#!/usr/bin/env python3
"""ManageSieve client"""

#
# Copyright 2023 and 2024  Odin Kroeger
#
# This file is part of SieveManager.
#
# SieveManager is free software: you can redistribute it and/or
# modify it under the terms of the GNU General Public License as
# published by the Free Software Foundation, either version 3 of
# the License, or (at your option) any later version.
#
# SieveManager is distributed in the hope that it will be useful,
# but WITHOUT ALL WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with SieveManager. If not, see <https://www.gnu.org/licenses/>.
#

#
# Metadata
#

__version__ = '0.6.1'
__author__ = 'Odin Kroeger'
__copyright__ = '2023 and 2024  Odin Kroeger'
__license__ = 'GPLv3+'
__all__ = [
    'SieveConnection',
    'SieveManager',
    'AbstractSASLChannel',
    'SieveSASLChannel',
    'AbstractAuth',
    'BaseAuth',
    'BasePassAuth',
    'BaseScramAuth',
    'BaseScramPlusAuth',
    'AuthorisationUnsupportedTrait',
    'CramMD5Auth',
    'ExternalAuth',
    'LoginAuth',
    'PlainAuth',
    'ExternalAuth',
    'ScramSHA1Auth',
    'ScramSHA1PlusAuth',
    'ScramSHA224Auth',
    'ScramSHA224PlusAuth',
    'ScramSHA256Auth',
    'ScramSHA256PlusAuth',
    'ScramSHA384Auth',
    'ScramSHA384PlusAuth',
    'ScramSHA512Auth',
    'ScramSHA512PlusAuth',
    'ScramSHA3_512Auth',
    'ScramSHA3_512PlusAuth',
    'Atom',
    'AuthPhase',
    'AuthType',
    'Backup',
    'Capabilities',
    'Response',
    'URL',
    'SASLPrep',
    'Error',
    'CertRevokedError',
    'CapabilityError',
    'AuthMechError',
    'ProxyAuthError',
    'TLSCapabilityError',
    'ConnectionOpenError',
    'ConnectionClosedError',
    'OCSPError',
    'OperationError',
    'OperationFailedError',
    'OperationOngoingError',
    'ProtocolError',
    'SocketError',
    'VerificationError'
]


#
# Modules
#

from abc import ABC, abstractmethod
from base64 import b64decode, b64encode
from collections import defaultdict
from collections.abc import Iterable
from dataclasses import dataclass
from errno import EEXIST, EISCONN, ENOENT, ENOTCONN
from fcntl import LOCK_SH, LOCK_NB
from getopt import GetoptError, getopt
from inspect import Parameter
from os import O_CREAT, O_EXCL, O_WRONLY, O_TRUNC, SEEK_END, SEEK_SET
from os import path
from signal import SIG_DFL, SIG_IGN, SIGHUP, SIGINT, SIGTERM
from tempfile import SpooledTemporaryFile, TemporaryDirectory
from typing import (Any, Callable, ClassVar, Final, Iterator,
                    Optional, Union, Sequence, TypeVar)

import code
import dataclasses
import datetime
import enum
import fcntl
import fnmatch
import getpass
import hashlib
import hmac
import inspect
import io
import ipaddress
import locale
import logging
import math
import netrc
import os
import platform
import pwd
import random
import re
import readline
import secrets
import select
import shlex
import shutil
import signal
import socket
import subprocess   # nosec B404
import ssl
import string
import stringprep
import sys
import threading
import traceback
import types
import unicodedata
import urllib.parse
import urllib.request

try:
    from cryptography import x509
    from cryptography.hazmat.primitives.hashes import SHA1
    from cryptography.hazmat.primitives.serialization import Encoding
    from cryptography.x509 import (AuthorityInformationAccess,
                                   ExtensionNotFound, ocsp)
    from cryptography.x509.oid import AuthorityInformationAccessOID
    from cryptography.x509.ocsp import OCSPResponseStatus, OCSPCertStatus
except ImportError:
    pass

try:
    import dns
    from dns.exception import DNSException
except ImportError:
    pass

try:
    import tkinter
except ImportError:
    pass


#
# Python version
#

if sys.version_info < (3, 9):
    sys.exit('SieveManager requires Python 3.9 or later')


#
# Types
#

class Atom(str):
    """ManageSieve keyword.

    .. seealso::
        * :RFC:`2244` (sec. 2.6.3)
        * :RFC:`5804` (secs. 1 and 4)
    """

    data: list['Word'] = []
    """Data attached to an atom."""


AuthMech = type['AbstractAuth']
"""SASL authentication mechanism."""


class AuthPhase(enum.IntEnum):
    """Phases of the authentication process."""

    PREAUTH = enum.auto()
    """"AUTHENTICATE" has *not* been issued."""

    SENT = enum.auto()
    """"AUTHENTICATE" issued, data sent, ready to receive."""

    RECEIVED = enum.auto()
    """"AUTHENTICATE" issued, data received, ready to send."""

    DONE = enum.auto()
    """Authentication concluded."""


class AuthType(enum.IntEnum):
    """Authentication type."""

    OTHER = enum.auto()
    PASSWORD = enum.auto()
    EXTERNAL = enum.auto()


class Backup(enum.IntEnum):
    """Emacs-like backup policies.

    .. seealso::
        * GNU `Emacs manual`_ (chap. 19.3.2)

    .. _`Emacs manual`: https://www.gnu.org/software/emacs/manual
    """

    NONE = 0
    """Do *not* make backups."""

    NUMBERED = enum.auto()
    """Back up :file:`file` to :file:`file.~{n}~`.

    *n* starts with 1 and increments with each backup.
    """

    SIMPLE = enum.auto()
    """Back up :file:`file` to :file:`file{~}`."""


class ConfirmEnum(enum.IntEnum):
    """Confirms an action."""

    NO = 0
    YES = 1
    ALL = 2
    NONE = 3

    def __bool__(self) -> bool:
        return self in (self.YES, self.ALL)


class LogLevel(enum.IntEnum):
    """Logging levels."""

    AUTH = logging.DEBUG - 10
    DEBUG = logging.DEBUG
    INFO = logging.INFO
    WARNING = logging.WARNING
    ERROR = logging.ERROR


class SASLPrep(enum.IntEnum):
    """Controls which strings are prepared for authentication.

    .. seealso::
        * :RFC:`4013`
        * :RFC:`4422` (sec. 4)
    """

    NONE = 0
    USERNAMES = 1
    PASSWORDS = 2
    ALL = 3


class ShellCmd(enum.IntEnum):
    """Shell actions that may overwrite or remove files."""

    NONE = 0
    CP = 1
    GET = 2
    MV = 4
    PUT = 8
    RM = 16
    ALL = 31


Word = Union[Atom, int, list, str]
"""ManageSieve word.

.. seealso::
    * :RFC:`2244` (sec. 2.6)
    * :RFC:`5804` (secs. 1.2 and 4)
"""


Line = list[Word]
"""ManageSieve line.

.. seealso::
    * :RFC:`2244` (sec. 2.2)
    * :RFC:`5804` (secs. 1.2 and 4)
"""

T = TypeVar('T')
"""Type variable."""


#
# Globals
#

# Directories
HOMEDIR: Final[str] = os.getenv('HOME', path.expanduser('~'))
"""Home directory."""

XDGCONFIGDIR: Final[str] = os.getenv('XDG_CONFIG_HOME', f'{HOMEDIR}/.config')
"""X Desktop group base configuration directory."""

# Debugging
DEBUG: bool = __debug__
"""Print stack traces even for expected error types?"""


#
# ManageSieve API
#

class SieveConnection():
    """Connection to a ManageSieve server.

    For example:

    >>> conn = SieveConnection('imap.foo.example', 'user', 'password')
    >>> with open('script.sieve', 'br') as file:
    >>>     res = conn.execute('PUTSCRIPT', file.name, file)
    >>>     if res.response != 'ok':
    >>>         raise res.toerror()
    >>> res = conn.execute('LOGOUT')
    >>> if res.response != 'ok':
    >>>     raise res.toerror()

    .. note::
        :class:`SieveConnect` provides a low-level interface.
        You likely want to use :class:`SieveManager` instead.

    .. warning::
        :class:`SieveConnection` is not thread-safe.

    .. see also::
        * :RFC:`2244` (ACAP)
        * :RFC:`2782` (DNS SRV)
        * :RFC:`5804` (ManageSieve)
    """

    def __init__(self, *args, **kwargs):
        """Connect to a ManageSieve server.

        Arguments:
            args: Positional arguments for :meth:`open`.
            kwargs: Keyword arguments for :meth:`open`.

        Raises:
            CertRevokedError: Server certificate was revoked.
            OCSPError: Server certificate status could not be checked.
            TLSCapabilityError: Server does not support TLS.
            ValueError: Bad characters in credentials.
        """
        self.lock = threading.Lock()
        self.open(*args, **kwargs)

    def __del__(self):
        """Shut the connection down."""
        try:
            self.shutdown()
        except OSError:
            pass

    # pylint: disable=too-many-branches,too-many-locals,too-many-nested-blocks
    def authenticate(self,
                     login: str,
                     *args,
                     owner: str = '',
                     sasl: Iterable[AuthMech] = (),
                     logauth: bool = False,
                     **kwargs):
        """Authenticate as `login`.

        Which credentials must be provided depends
        on the given SASL mechanisms.

        Password-based mechanisms require a password:

        >>> conn.authenticate('user', 'password')

        By contrast, the "EXTERNAL" mechanism requires no credentials:

        >>> conn.authenticate('user', sasl=[ExternalAuth])

        By default, authentication is attempted with password-based
        mechanisms, starting with those with better security properties
        and progressing to those with worse security properties.

        Mixing authentication mechanisms that take different
        credentials begs for a :exc:`TypeError`.

        Arguments:
            login: User to login as (authentication ID).
            args: Positional arguments for SASL mechanism constructors.
            owner: User whose scripts to manage (authorisation ID).
            sasl: SASL mechanisms to try
                (default: non-obsolete subclasses of :class:`BasePassAuth`).
            logauth: How to log the authentication exchange.
            kwargs: Keyword arguments for SASL mechanism constructors.

        Raises:
            AuthMechError: SASL mechanisms not supported by server.
            OperationFailedError: Authentication failed.
            ProxyAuthError: Authorisation unsupported.
            ValueError: Bad characters in credentials.
        """
        kwargs['authzid'] = owner
        logger = self.logger
        self.auth = args
        self.kwauth = kwargs
        self.sasl = sasl if sasl else BasePassAuth.subclasses()
        self.logauth = logauth
        while True:
            # pylint: disable=consider-using-with
            if not self.lock.acquire(blocking=False):
                raise OperationOngoingError('other operation ongoing')
            try:
                for mech in self.sasl:
                    assert self.capabilities    # nosec B101
                    if mech.name.casefold() in self.capabilities.sasl:
                        conn = SieveSASLChannel(self, log=logauth)
                        try:
                            auth = mech(conn, login, *args, **kwargs)
                            if caps := auth():
                                self.capabilities = caps
                        except CapabilityError as err:
                            logger.info(err)
                            continue
                        except OperationFailedError as err:
                            # TRANSITION-NEEDED does not require special
                            # treatment. Either the user can be authenticated
                            # with a subsequent mechanism (e.g., SCRAM-SHA-1
                            # after SCRAM-SHA-256) or a plaintext mechanism
                            # should be reached eventually.
                            if err.matches('auth-too-weak',
                                           'encrypt-needed',
                                           'transition-needed'):
                                continue
                            raise err
                        if login := auth.authcid:
                            self.login = login
                            logger.info('authenticated as %s', login)
                        if owner := auth.authzid:
                            self.owner = owner
                            logger.info('authorised as %s', owner)
                        return
                raise AuthMechError('authentication mechanisms exhausted')
            except ConnectionClosedError as err:
                if err.matches('referral'):
                    try:
                        # pylint: disable=unbalanced-tuple-unpacking
                        url, = err.data
                    except ValueError as valuerr:
                        raise ProtocolError('unexpected data') from valuerr
                    if isinstance(url, Atom) or not isinstance(url, str):
                        # pylint: disable=raise-missing-from
                        raise ProtocolError('expected string')
                    self._followreferral(url)
                    continue
                raise err
            finally:
                self.lock.release()
            # NOTREACHED

    def close(self):
        """Close the client-side of the connection.

        .. warning::
            Should only be called if the server has closed the connection.
        """
        if self.file is not None:
            self.file.close()
            self.file = None
        if self.poll is not None:
            assert self.sock     # nosec B101
            self.poll.unregister(self.sock)
            self.poll = None
        if self.sock is not None:
            self.sock.close()
            self.sock = None
        self.capabilities = None
        self.host = None
        self.port = None
        self.login = ''
        self.owner = ''
        self.auth = ()
        self.kwauth = {}

    def collect(self, check: bool = False) -> tuple['Response', list[Line]]:
        """Collect the server's response to the last transmission.

        For example:

        >>> conn.sendline(Atom('LISTSCRIPTS'))
        >>> conn.collect()
        (Response(response='ok', code=(), data=(),
         message='Listscripts completed'),
         [['foo.sieve', 'ACTIVE'], ['bar.sieve'], ['baz.sieve']])

        Arguments:
            check: Raise an error if the response is not "OK"?

        Returns:
            A :class:`Response` object and a list of ACAP lines,
            where each line is a list of ACAP words.
        """
        lines: list[Line] = []
        while True:
            words = self.receiveline()
            if words and isinstance(words[0], Atom):
                res = Response.fromdata(words)
                if check and res.response != 'ok':
                    raise res.toerror()
                if res.matches('warnings') and res.message:
                    self.warning = res.message
                    for line in res.message.splitlines():
                        self.logger.warning('%s', escapectrl(line))
                else:
                    self.warning = ''
                return res, lines
            lines.append(words)
        # NOTREACHED

    def execute(self, command: str,
                *args: Union[Atom, io.IOBase, bytes, str, int]) \
                    -> tuple['Response', list[Line]]:
        """Execute `command` and return the server's response.

        For example:

            >>> conn.execute('LISTSCRIPTS')
            (Response(response='ok', code=(), data=(),
             message='Listscripts completed'),
             [['foo.sieve', 'ACTIVE'], ['bar.sieve'], ['baz.sieve']])

        Returns:
            A list of ACAP lines, where each line is a list of ACAP words.

        .. note::
            Referrals are followed automatically.
        """
        assert command      # nosec B101
        while True:
            # pylint: disable=consider-using-with
            if not self.lock.acquire(blocking=False):
                raise OperationOngoingError('other operation ongoing')
            try:
                self.sendline(Atom(command.upper()), *args)
                res, data = self.collect()
            finally:
                self.lock.release()
            if res.response == 'ok':
                return res, data
            if res.response == 'no':
                raise res.toerror()
            if res.response == 'bye':
                if not res.matches('referral'):
                    raise res.toerror()
                try:
                    # pylint: disable=unbalanced-tuple-unpacking
                    url, = res.data
                except ValueError as valuerr:
                    raise ProtocolError('unexpected data') from valuerr
                if isinstance(url, Atom) or not isinstance(url, str):
                    # pylint: disable=raise-missing-from
                    raise ProtocolError('expected string')
                self._followreferral(url)
            else:
                raise ProtocolError(f'unexpected response: {res}')
        # NOTREACHED

    def memento(self) -> tuple[list, dict[str, Any]]:
        """Get arguments that would :meth:`open` the current connection.

        For example:

        >>> conn.host
        imap.foo.example
        >>> conn.login
        user
        >>> args, kwargs = conn.memento()
        >>> conn.shutdown()
        >>> conn.host
        None
        >>> conn.login
        None
        >>> conn.open(*args, **kwargs)
        >>> conn.host
        imap.foo.example
        >>> conn.login
        user

        """
        args = []
        kwargs = {}
        for name, param in list(self._signature.parameters.items())[1:]:
            value = getattr(self, name)
            if param.kind in (Parameter.POSITIONAL_ONLY,
                                Parameter.POSITIONAL_OR_KEYWORD):
                args.append(value)
            elif param.kind == Parameter.VAR_POSITIONAL:
                args.extend(value)
            elif param.kind == Parameter.KEYWORD_ONLY:
                kwargs[name] = value
            elif param.kind == Parameter.VAR_KEYWORD:
                kwargs.update(value)
        return (args, kwargs)

    # pylint: disable=redefined-outer-name
    def open(self,
             host: str,
             login: str,
             *auth,
             port: int = 4190,
             source: tuple[str, int] = ('', 0),
             timeout: Optional[float] = socket.getdefaulttimeout(),
             tls: bool = True,
             ocsp: bool = True,
             owner: str = '',
             sasl: Iterable[AuthMech] = (),
             logauth: bool = False,
             **kwauth):
        """Connect to `host` and login as `login`.

        Connect and login:

        >>> conn.open('imap.foo.example', 'user', 'password')

        Defer login:

        >>> conn.open('imap.foo.example', '')
        >>> conn.authenticate('user', 'password')

        Arguments:
            host: Server address.
            login: User to login as (authentication ID).
                Set to the empty string to defer login.
            auth: Positional arguments for :meth:`authenticate`.
            port: Server port.
            source: Source address and port.
            timeout: Timeout in seconds.
            tls: Secure the connection?
            ocsp: Check whether the server certificate was revoked? [#ocsp]_
            owner: User whose scripts to manage (authorisation ID).
            sasl: See :meth:`authenticate`.
            logauth: See :meth:`authenticate`.
            kwauth: Keywords arguments for :meth:`authenticate`.

        Raises:
            AuthMechError: SASL mechanisms not supported by server.
            CertRevokedError: Server certificate was revoked.
            ConnectionOpenError: Already connected.
            OCSPError: Server certificate status could not be checked.
            ProxyAuthError: Authorisation unsupported.
            TLSCapabilityError: Server does not support TLS.
            OperationFailedError: Authentication failed.
            ValueError: Bad characters in credentials.

        .. [#ocsp] Silently ignored unless the non-standard Python
                   module cryptography_ has been installed.

        .. _cryptography: https://cryptography.io.
        """
        # pylint: disable=consider-using-with
        if not self.lock.acquire(blocking=False):
            raise OperationOngoingError('other operation ongoing')
        try:
            if self.sock or self.file:
                raise ConnectionOpenError(EISCONN, os.strerror(EISCONN))
            try:
                if 'dns.resolver' not in sys.modules:
                    raise ResolutionError('dnspython unavailable')
                if isipaddress(host):
                    raise ResolutionError('not a hostname')
                # This is the algorithm specified by RFC 2782,
                # NOT the one specified by RFC 5804, sec. 1.8,
                # because the latter is wrong.
                try:
                    for rec in resolvesrv(f'_sieve._tcp.{host}.'):
                        try:
                            self._connect(rec.host, rec.port, source, timeout)
                        except OSError:
                            continue
                        break
                except DNSException as err:
                    # DNSException could raise a NameError below.
                    raise ResolutionError('host not found') from err
            except ResolutionError:
                self._connect(host, port, source, timeout)
            _, data = self.collect(check=True)
            self.capabilities = Capabilities.fromdata(data)
            self.source = source
            self.logger.info('connected to %s:%d', host, port)
        finally:
            self.lock.release()
        if tls:
            self._starttls(ocsp=ocsp)
        if login:
            self.authenticate(login, *auth, owner=owner, sasl=sasl,
                              logauth=logauth, **kwauth)

    def receiveline(self, log: bool = True) -> Line:
        """Receive a line from the server and parse it.

        The ManageSieve protocol uses the same syntax and data types
        as the Application Configuration Access Protocol (ACAP).

        ==================  =============
        ACAP type           Python type
        ==================  =============
        Atom                :class:`Atom`
        Literal             `str`
        Number              `int`
        Parenthesised List  `list`
        String              `str`
        ==================  =============

        SieveManager converts the ACAP type "NIL" to an :class:`Atom`,
        because ManageSieve does not incorporate that type.

        For example:

        >>> conn.sendline(Atom('LISTSCRIPTS'))
        >>> conn.receiveline()
        ['foo.sieve', 'ACTIVE']
        >>> conn.receiveline()
        ['bar.sieve']
        >>> conn.receiveline()
        ['baz.sieve']
        >>> conn.receiveline()
        ['OK', 'Listscripts completed.']

        Arguments:
            log: Log received lines with priority :attr:`logging.DEBUG`?

        Returns:
            A list of ACAP words that comprise an ACAP line.

        .. seealso::
            * :RFC:`2244` (sec. 2.6)
            * :RFC:`5804` (sec. 1.2)
        """
        assert self.file   # nosec B101
        words: Line = []
        stack: list[Line] = [words]
        ptr: Line = words
        self._checksocket()
        while line := self.file.readline().decode('utf8'):
            size: int = -1
            if log:
                self.logger.debug('S: %s', escapectrl(line.rstrip()))
            for token in self._acapregex.finditer(line):
                key = token.lastgroup
                value = token.group(key)    # type: ignore[arg-type]
                if key == 'leftparen':
                    parens: list[Word] = []
                    stack.append(ptr)
                    ptr.append(parens)
                    ptr = parens
                elif key == 'rightparen':
                    ptr = stack.pop()
                elif key == 'atom':
                    ptr.append(Atom(value))
                elif key == 'number':
                    ptr.append(int(value))
                elif key == 'string':
                    ptr.append(value)
                elif key == 'literal':
                    size = int(value)
                    literal = self.file.read(size).decode('utf8')
                    ptr.append(literal)
                    if log and self.logger.isEnabledFor(logging.DEBUG):
                        for logent in literal.splitlines():
                            self.logger.debug('S: %s', escapectrl(logent))
            if size == -1:
                break
        return words

    def sendline(self,
                 *objs: Union[Atom, io.IOBase, bytes, int, str],
                 log: bool = True):
        """Convert objects to ACAP types and send them as a single line.

        The ManageSieve protocol uses the same syntax and data types
        as the Application Configuration Access Protocol (ACAP).

        ==================  =============================
        Python type         ACAP type
        ==================  =============================
        :class:`Atom`       Atom
        :class:`io.IOBase`  Literal
        `bytes`             Literal or String [#literal]_
        `int`               Number
        `str`               Literal or String [#literal]_
        ==================  =============================

        SieveManager does not support converting to the ACAP types
        "Parenthesised List" or "NIL", because no ManageSieve command
        takes arguments of those types.

        For example:

        >>> conn.sendline(Atom('HAVESPACE'), 'script.sieve', 12345)
        >>> conn.collect()
        (Response(response='ok', code=(), data=[],
         message='Putscript would succeed'), [])

        Pipeline commands:

        >>> mgr.sendline(Atom('PUTSCRIPT', script, script.name))
        >>> mgr.sendline(Atom('LOGOUT'))
        >>> for _ in range(2):
        >>>     mgr.collect(check=True)

        .. [#literal] Depending on content.

        Arguments:
            log: Log sent lines with priority :attr:`logging.DEBUG`?

        .. seealso::
            * :RFC:`2244` (sec. 2.6)
            * :RFC:`5804` (sec. 1.2)
        """
        assert self.sock     # nosec B101
        buffer = bytearray()

        def encode(data: str) -> bytes:
            return unicodedata.normalize('NFC', data).encode('utf8')

        def send():
            self._checksocket()
            self.sock.sendall(buffer)    # type: ignore[union-attr]
            if log and self.logger.isEnabledFor(logging.DEBUG):
                for logent in buffer.decode('utf8').splitlines():
                    self.logger.debug('C: %s', escapectrl(logent))

        def addbytes(data: bytes):
            if self._stringregex.fullmatch(data):
                buffer.extend(b'"%s"' % data)
            else:
                buffer.extend(b'{%d+}\r\n%s' % (len(data), data))

        for i, obj in enumerate(objs):
            if i > 0:
                buffer.extend(b' ')
            if isinstance(obj, Atom):
                buffer.extend(encode(obj))
            elif isinstance(obj, int):
                buffer.extend(encode(str(obj)))
            elif isinstance(obj, bytes):
                addbytes(obj)
            elif isinstance(obj, str):
                addbytes(encode(obj))
            elif isinstance(obj, (io.IOBase, SpooledTemporaryFile)):
                buffer.extend(b'{%d+}\r\n' % getfilesize(obj))
                send()
                buffer = bytearray()
                try:
                    self.sock.sendfile(obj)  # type: ignore[attr-defined]
                except ValueError:
                    while block := encode(obj.read(io.DEFAULT_BUFFER_SIZE)):
                        self.sock.sendall(block)
                if log:
                    self.logger.debug('C: <file>')
            else:
                raise TypeError(type(obj).__name__ + ': not an ACAP type')
        buffer.extend(b'\r\n')
        send()

    def shutdown(self):
        """Shut the connection down.

        .. note::
            Should only be used if logging out would be unsafe.
        """
        if self.sock is not None:
            self.sock.shutdown(socket.SHUT_RDWR)
            self.logger.info('shut connection down')
        self.close()

    def geturl(self) -> Optional['URL']:
        """URL of the current connection.

        For example:

        >>> conn = SieveConnection('imap.foo.example', 'user', 'password')
        >>> conn.geturl()
        'sieve://user@imap.foo.example'

        .. note::
            Changes to the state of the connection that are effected by
            :meth:`sendline` or :meth:`execute` are *not* tracked.
        """
        if self.host:
            return URL(hostname=self.host,
                       port=self.port if self.port != 4190 else None,
                       username=self.login, owner=self.owner)
        return None

    @property
    def timeout(self) -> Optional[float]:
        """Connection timeout in seconds.

        Set timeout to 500 ms:

        >>> conn.timeout = 0.5

        .. note::
            The timeout can only be set while a connection is open.
        """
        if self.sock:
            return self.sock.gettimeout()
        return None

    @timeout.setter
    def timeout(self, secs: Optional[float]):
        if self.sock:
            self.sock.settimeout(secs)

    @property
    def tls(self) -> Optional[str]:     # type: ignore[override]
        """TLS version."""
        if isinstance(self.sock, ssl.SSLSocket):
            return self.sock.version()
        return None

    auth: tuple = ()
    """Positional arguments for SASL mechanism constructors."""

    capabilities: Optional['Capabilities'] = None
    """Server capabilities."""

    file: Optional[io.IOBase] = None
    """File-like access to :attr:`sock`."""

    host: Optional[str] = None
    """Remote address."""

    kwauth: dict[str, Any] = {}
    """Keyword arguments for SASL mechanism constructors."""

    lock: threading.Lock
    """Operation lock."""

    logauth: bool
    """Log the authentication exchange?"""

    logger: logging.Logger = logging.getLogger(__name__)
    """Logger to use.

    :class:`SieveConnection` logs messages of the following priorities:

    ========================  =====================================
    Priority                  Used for
    ========================  =====================================
    :const:`logging.ERROR`    Non-fatal errors
    :const:`logging.WARNING`  Server warnings
    :const:`logging.INFO`     State changes
    :const:`logging.DEBUG`    Data sent to/received from the server
    ========================  =====================================

    Suppress logging:

    >>> from logging import getLogger
    >>> getLogger('sievemgr').setLevel(logging.CRITICAL)

    Use an individual logger:

    >>> from logging import getLogger
    >>> conn.logger = getLogger('foo').addHandler(logging.NullHandler())

    Print data send to/received from the server to standard error:

    >>> from logging import getLogger
    >>> getLogger('sievemgr').setLevel(logging.DEBUG)
    >>> conn.execute('LISTSCRIPTS')
    C: LISTSCRIPTS
    S: "foo.sieve" ACTIVE
    S: "bar.sieve"
    S: "baz.sieve"
    S: OK "Listscripts completed"
    (Response(response='ok', code=(), data=(),
     message='Listscripts completed'),
     [('foo.sieve', True), ('bar.sieve', False), ('baz.sieve', False)])
    """

    login: str = ''
    """Login name (authentication ID)."""

    ocsp: bool
    """Check whether the server certificate was revoked? [#ocsp]_"""

    owner: str = ''
    """User whose scripts are managed (authorisation ID)."""

    poll: Optional[select.poll] = None
    """Polling object for :attr:`sock`."""

    port: Optional[int] = None
    """Remote port."""

    sasl: Iterable[AuthMech] = ()
    """SASL mechanisms."""

    sock: Optional[socket.SocketType] = None
    """Underlying socket."""

    source: tuple[str, int] = ('', 0)
    """Source address and port."""

    sslcontext: ssl.SSLContext = ssl.create_default_context()
    """Settings for negotiating Transport Layer Security (TLS).

    Disable workarounds for broken X.509 certificates:

    >>> SieveConnection.sslcontext.verify_flags |= ssl.VERIFY_X509_STRICT

    Load client certificate/key pair:

    >>> SieveConnection.sslcontext.load_cert_chain(cert='cert.pem')

    Use a custom certificate authority:

    >>> from ssl import create_default_context
    >>> SieveConnection.sslcontext = create_default_context(cafile='site.pem')

    """

    warning: str = ''
    """Warning issued in response to last "CHECKSCRIPT" or "PUTSCRIPT".

    For example:

    >>> with open('script.sieve', 'br') as file:
    >>>     conn.execute('PUTSCRIPT', file, file.name)
    (Response(response='ok', code=('warnings,'),
     data=('line 7: may need to be frobnicated',),
     message='Putscript succeeded'), [])
    >>> conn.warning
    'line 7: may need to be frobnicated'

    .. note::
        Only set by :meth:`collect`, :meth:`execute`,
        and :class:`SieveManager` methods.

    .. seealso::
        * :RFC:`5804` (secs. 2.6 and 2.12)
    """

    def _checksocket(self):
        """Check whether the socket is alive.

        Raises:
            SocketError: Socket error.
        """
        assert self.poll   # nosec B101
        for _, events in self.poll.poll():
            if events & select.POLLERR:
                raise SocketError(ENOTCONN, 'socket error')
            if events & select.POLLHUP:
                raise SocketError(ENOTCONN, 'server hung up')
            if events & select.POLLNVAL:
                raise SocketError(ENOTCONN, os.strerror(ENOTCONN))

    def _connect(self,
                 host: str,
                 port: int = 4190,
                 source: tuple[str, int] = ('', 0),
                 timeout: Optional[float] = socket.getdefaulttimeout()):
        """Connect to `host` at `port`."""
        self.sock = socket.create_connection((host, port), timeout=timeout,
                                                source_address=source)
        self.file = self.sock.makefile('rb')
        self.poll = select.poll()
        self.poll.register(self.sock)
        self.host = host
        self.port = port

    def _followreferral(self, url: str):
        """Close the current connection and :meth:`open` `url` instead."""
        try:
            ref = URL.fromstr(url)
        except ValueError as err:
            raise ProtocolError(err) from err
        self.logger.info('referred to %s', url)
        args, kwargs = self.memento()
        del args[0]
        del kwargs['port']
        host = ref.hostname
        port = 4190 if ref.port is None else ref.port
        self.close()
        self.open(host, *args, port=port, **kwargs)

    # pylint: disable=redefined-outer-name
    def _starttls(self, ocsp: bool = True):
        """Start TLS encryption.

        Arguments:
            ocsp: Check whether the server certificate was revoked?

        Raises:
            CertRevokedError: Server certificate was revoked.
            OCSPError: Server certificate status could not be checked.
            TLSCapabilityError: Server does not support TLS.

        .. seealso::
            * :func:`ssl.SSLContext.wrap_socket`
        """
        assert self.sock            # nosec B101
        assert self.capabilities    # nosec B101
        if not self.capabilities.starttls:
            raise TLSCapabilityError('STARTTLS: not supported')
        self.execute('STARTTLS')
        # pylint: disable=consider-using-with
        if not self.lock.acquire(blocking=False):
            raise OperationOngoingError('other operation ongoing')
        try:
            self.sock = self.sslcontext.wrap_socket(
                self.sock,  # type: ignore[arg-type]
                server_hostname=self.host
            )
            self.file = self.sock.makefile('rb')
            _, data = self.collect(check=True)
            self.capabilities = Capabilities.fromdata(data)
            proto = self.sock.version()
            cipher = self.sock.cipher()
            if proto and cipher:
                self.logger.info('started %s using %s', proto, cipher[0])
            if ocsp:
                if 'cryptography' in sys.modules:
                    der = self.sock.getpeercert(binary_form=True)
                    cert = x509.load_der_x509_certificate(der)
                    if certrevoked(cert):
                        raise CertRevokedError('server certificate revoked')
                else:
                    self.logger.error('cryptography module not found, '
                                      'certificate status not checked')
            self.ocsp = ocsp
        finally:
            self.lock.release()

    _acapregex: re.Pattern = re.compile('|'.join((
        r'(?P<atom>[a-z][^(){\s\\]*)',
        r'(?P<number>\d+)',
        r'"(?P<string>[^\0\r\n"]*)"',
        r'{(?P<literal>\d+)\+?}',
        r'(?P<leftparen>\()',
        r'(?P<rightparen>\))',
    )), flags=re.IGNORECASE)
    """Regular expression to parse an ACAP line.

    Should be used with :meth:`re.itermatch`.

    .. seealso::
        * :RFC:`2244` (sec. 2.6.1)
        * :RFC:`5804` (sec. 1.2)
    """

    # Strings may be 1024 octets long, but I limit the length to 1020 octets
    # to allow for errors (two octets for quotations marks, one for the
    # terminating null byte, one for other off-by-one errors).
    _stringregex: re.Pattern = re.compile(br'[^\0\r\n"]{0,1020}')
    """Controls whether a `str` is converted into an ACAP string or literal.

    Must be used with :meth:`re.fullmatch`.

    .. seealso::
        * :RFC:`2244` (sec. 2.6.3)
        * :RFC:`5804` (sec. 1.2)
    """

    _signature = inspect.signature(open)
    """Signature of :meth:`open`."""


class SieveManager(SieveConnection):
    """High-level interface to a ManageSieve connection.

    For example:

        >>> with SieveManager('imap.foo.example', 'user', 'password') as mgr:
        >>>     with open('sieve.script', 'br') as script:
        >>>         mgr.putscript(script, script.name)
        >>>     mgr.setactive('sieve.script')

    .. note::
        :class:`SieveManager` should *always* be used as a context manager.

    .. warning::
        :class:`SieveManager` is not thread-safe.
    """

    def __init__(self, *args, backup: Backup = Backup.NONE, **kwargs):
        """Connect to a ManageSieve server.

        Arguments:
            args: Positional arguments for :meth:`SieveConnection.open`.
            backup: Default backup policy.
            kwargs: Keyword arguments for :meth:`SieveConnection.open`.

        Raises:
            OCSPError: Server certificate status could not be checked.
            TLSCapabilityError: Server does not support TLS.
            ValueError: Bad characters in credentials.
        """
        super().__init__(*args, **kwargs)
        self.backup: Backup = backup

    def __enter__(self) -> 'SieveManager':
        """Enter a context."""
        return self

    # pylint: disable=redefined-outer-name
    def __exit__(self, exc_type, exc_value, traceback):
        """Exit the context and close the connection appropriately."""
        if isinstance(exc_value, (ConnectionError, TimeoutError)):
            self.close()
        elif isinstance(exc_value, ProtocolError):
            self.shutdown()
        else:
            try:
                self.logout()
            except OperationOngoingError:
                try:
                    self.shutdown()
                except OSError:
                    self.close()

    def backupscript(self, script: str,
                     policy: Backup = Backup.SIMPLE,
                     checkspace: int = 262_144,
                     memory: int = 524_288):
        """Make a backup of `script`.

        Arguments:
            script: Script name.
            policy: Backup policy.
            checkspace: See :meth:`putscript`.
            memory: See :meth:`copyscript`.
        """
        def getscripts() -> tuple[str, ...]:
            return tuple((s for s, _ in self.listscripts()))
        if backup := getbackupname(script, getscripts, policy):
            self.copyscript(script, backup, backup=Backup.NONE,
                            checkspace=checkspace, memory=memory)

    def capability(self) -> 'Capabilities':
        """Get the server's capabilities."""
        try:
            return Capabilities.fromdata(self.execute('CAPABILITY')[1])
        except ValueError as err:
            raise ProtocolError('expected words') from err

    def checkscript(self, script: Union[str, io.IOBase]):
        """Check whether `script` is semantically valid.

        Arguments:
            script: Script (*not* script name)
        """
        assert self.capabilities    # nosec B101
        if not self.capabilities.version:
            raise CapabilityError('CHECKSCRIPT: not supported')
        self.execute('CHECKSCRIPT', script)

    def copyscript(self,
                   source: str,
                   target: str,
                   backup: Optional[Backup] = None,
                   checkspace: int = 262_144,
                   memory: int = 524_288):
        """Download `source` and re-upload it as `target`.

        Arguments:
            source: Source name.
            target: Target name.
            backup: Backup policy (default: :attr:`backup`).
            checkspace: See :meth:`putscript`.
            memory: Memory to allocate for `source` in bytes.
                If `source` is larger, it is written to a temporary file.
        """
        with SpooledTemporaryFile(max_size=memory, mode='bw+') as temp:
            temp.write(self.getscript(source).encode('utf8'))
            temp.seek(0)
            self.putscript(temp, target, backup=backup, checkspace=checkspace)

    def deletescript(self, script: str):
        """Delete `script`."""
        self._checkscriptname(script)
        self._scripts = None
        self.execute('DELETESCRIPT', script)
        self.logger.info('removed %s', script)

    # pylint: disable=too-many-locals,too-many-branches
    def editscripts(self,
                    command: list[str],
                    scripts: list[str],
                    *args,
                    check: bool = True,
                    create: bool = True,
                    encoding: str = locale.getpreferredencoding(),
                    **kwargs) -> subprocess.CompletedProcess:
        """Download `scripts`, modify them, and re-upload them.

        Script names are appended to `command`.
        If the server disconnects while `command` is running,
        the connection is re-established auto-magically.
        Only scripts that were modified are re-uploaded.

        For example:

        >>> mgr.editscripts(['vi'], ['foo.sieve'])

        >>> cp = mgr.editscripts(['cmp'], ['foo.sieve', 'bar.sieve'])
        >>> if cp.returncode != 0:
        >>>     print('foo.sieve and bar.sieve differ')

        Arguments:
            command: Command to run.
            scripts: Script names.
            check: See :func:`subprocess.run`.
            create: Create scripts that do not exist?
            encoding: File encoding.
            args: Passed to :func:`subprocess.run` as positional arguments.
            kwargs: Passed to :func:`subprocess.run` as keyword arguments.

        Raises:
            ValueError: Script name contains a path separator.

        .. seealso::
            * :func:`subprocess.run`
        """
        oargs, okwargs = self.memento()
        for script in scripts:
            self._checkscriptname(script)
            if path.sep in script:
                raise ValueError(f'{script}: contains {path.sep}')
        with TemporaryDirectory() as tmpdir:
            fnames = []
            ctimes = []
            for script in scripts:
                fname = path.join(tmpdir, script)
                with open(fname, 'w', encoding=encoding) as file:
                    try:
                        file.write(self.getscript(script))
                    except OperationFailedError as err:
                        if not create:
                            raise err
                        if not err.code:
                            for existing, _ in self.listscripts():
                                if existing == script:
                                    raise err
                        elif not err.matches('nonexistent'):
                            raise err
                fnames.append(fname)
                ctimes.append(os.stat(fname).st_ctime)
            retval = subprocess.run(command + fnames, *args,    # nosec B603
                                    check=check, **kwargs)
            for script, fname, ctime in zip(scripts, fnames, ctimes):
                if os.stat(fname).st_ctime > ctime:
                    for _ in range(2):
                        try:
                            with open(fname, 'rb') as file:
                                self.putscript(file, script)
                            break
                        except ConnectionError:
                            self.close()
                            self.open(*oargs, **okwargs)
        return retval

    def getactive(self) -> Optional[str]:
        """Get the name of the active script."""
        for name, active in self.listscripts():
            if active:
                return name
        return None

    def getscript(self, script: str) -> str:
        """Download `script`.

        For example:

        >>> with open('foo.sieve', 'bw') as file:
        >>>     file.write(mgr.getscript(file.name))

        Arguments:
            script: Script name.
        """
        self._checkscriptname(script)
        try:
            # pylint: disable=unbalanced-tuple-unpacking
            _, ((content,),) = self.execute('GETSCRIPT', script)
        except ValueError as err:
            raise ProtocolError('unexpected data') from err
        if isinstance(content, Atom):
            raise ProtocolError('unexpected atom')
        if isinstance(content, str):
            return content
        raise ProtocolError(f'unexpected {type(content).__name__}')

    def havespace(self, script: str, size: int):
        """Check whether there is enough space for `script`.

        Arguments:
            script: Script name (*not* script).
            size: Script size in bytes.

        Raises:
            OperationFailedError: There is *not* enough space.
            ValueError: `size` is negative.
        """
        self._checkscriptname(script)
        if size < 0:
            raise ValueError('size must be non-negative')
        self.execute('HAVESPACE', script, size)

    @classmethod
    def isscriptname(cls, scriptname: str) -> bool:
        """Check whether `scriptname` is a valid script name."""
        return bool(cls._scriptnameregex.fullmatch(scriptname))

    def listscripts(self, cached: bool = False) -> list[tuple[str, bool]]:
        """List scripts and whether they are the active script.

        For example:

        >>> mgr.listscripts()
        [('foo.sieve', False), ('bar.sieve', True)]

        >>> scripts = [script for script, _ in mgr.listscripts()]

        Arguments:
            cached: Return cached response? [#cache]_

        Returns:
            A list of script name/status tuples.

        .. [#cache]  The cache is emptied when a
                     state-changing operation is requested.
        """
        if not cached or self._scripts is None:
            self._scripts = []
            for line in self.execute('LISTSCRIPTS')[1]:
                try:
                    name = line[0]
                except IndexError as err:
                    raise ProtocolError('no data') from err
                if not isinstance(name, str):
                    raise ProtocolError('expected string')
                try:
                    status = line[1]
                    if not isinstance(status, str):
                        raise ProtocolError('expected string')
                    active = status.casefold() == 'active'
                except IndexError:
                    active = False
                self._scripts.append((name, active))
        return self._scripts

    def logout(self):
        """Log out.

        .. note::
            :meth:`logout` should be called to close the connection
            unless :class:`SieveManager` is used as a context manager.

        .. warning::
            Logging out is unsafe after a :exc:`ProtocolError`.
            Use :meth:`shutdown` instead.
        """
        if self.sock is not None:
            try:
                self.execute('LOGOUT')
                self.logger.info('logged out')
            except (OperationError, ProtocolError):
                try:
                    self.shutdown()
                except OSError:
                    pass
            except ConnectionError:
                pass
        self.close()

    def noop(self, tag: Optional[str] = None) -> Optional[str]:
        """Request a no-op.

        For example:

        >>> mgr.noop('foo')
        'foo'

        Arguments:
            tag: String for the server to echo back.

        Returns:
            Server echo.

        Raises:
            CapabilityError: Server does not support "NOOP".
        """
        assert self.capabilities    # nosec B101
        if not self.capabilities.version:
            raise CapabilityError('NOOP: not supported')
        args = () if tag is None else (tag,)
        res, _ = self.execute('NOOP', *args)
        if not res.data:
            return None
        try:
            # pylint: disable=unbalanced-tuple-unpacking
            data, = res.data
        except ValueError as err:
            raise ProtocolError('unexpected words') from err
        if isinstance(data, Atom) or not isinstance(data, str):
            raise ProtocolError('expected string')
        return data

    def putscript(self,
                  source: Union[str, io.IOBase],
                  target: str,
                  backup: Optional[Backup] = None,
                  checkspace: int = 262_144):
        """Upload `source` to the server as `target`.

        For example:

        >>> mgr.putscript('# empty', 'foo.sieve')

        >>> with open('foo.sieve', 'br') as file:
        >>>     mgr.putscript(file, 'foo.sieve')

        Arguments:
            source: Script.
            target: Script name.
            backup: Backup policy (default: :attr:`backup`).
            checkspace: Check whether there is enough space
                if `script`'s size is greater than the given size in bytes.
                Set to 0 to always check and to -1 to never check.
        """
        self._checkscriptname(target)
        if checkspace > -1:
            if isinstance(source, str):
                size = len(source)
            elif isinstance(source, (io.IOBase, SpooledTemporaryFile)):
                size = getfilesize(source)
            else:
                raise TypeError('source must be str or file-like object')
            if size > checkspace:
                self.havespace(target, size)
        try:
            policy = self.backup if backup is None else backup
            self.backupscript(target, policy=policy)
        except OperationFailedError as err:
            if not err.code:
                for script, _ in self.listscripts():
                    if script == target:
                        raise err
            elif not err.matches('nonexistent'):
                raise err
        self._scripts = None
        self.execute('PUTSCRIPT', target, source)
        self.logger.info('uploaded %s', target)

    def renamescript(self, source: str, target: str,
                     emulate: bool = True, checkspace: int = 262_144,
                     memory: int = 524_288):
        """Rename `source` to `target`.

        Some servers do not the support the "RENAMESCRIPT" command.
        On such servers, renaming may be emulated by downloading `source`,
        uploading it as `target`, setting `target` to the active script
        if `source` is the active script, and then deleting `source`.

        For example:

        >>> mgr.renamescript('foo.sieve', 'bar.sieve', emulate=False)

        Arguments:
            source: Script name.
            target: Script name.
            emulate: Emulate "RENAMESCRIPT" if the server does not support it?
            checkspace: See :meth:`putscript`.
            memory: See :meth:`copyscript`.

        Raises:
            CapabilityError: `emulate` is `False` and the
                server does not support "RENAMESCRIPT".
            OperationFailedError: `source` does not exist or `target` exists.
        """
        assert self.capabilities    # nosec B101
        self._checkscriptname(source)
        self._checkscriptname(target)
        if self.capabilities.version:
            self._scripts = None
            self.execute('RENAMESCRIPT', source, target)
            self.logger.info('renamed %s to %s', source, target)
        elif emulate:
            sourceactive: Optional[bool] = None
            for script, active in self.listscripts():
                if script == source:
                    sourceactive = active
                if script == target:
                    message = os.strerror(EEXIST)
                    raise OperationFailedError(code='alreadyexists',
                                               message=f'{target}: {message}')
            if sourceactive is None:
                message = os.strerror(ENOENT)
                raise OperationFailedError(code='nonexistent',
                                           message=f'{source}: {message}')
            self.copyscript(source, target, backup=Backup.NONE,
                            checkspace=checkspace, memory=memory)
            if sourceactive:
                self.setactive(target)
            self.deletescript(source)
        else:
            raise CapabilityError('RENAMESCRIPT: not supported')

    def scriptexists(self, script: str, cached: bool = False) -> bool:
        """Check if `script` exists.

        Arguments:
            cached: Return cached response? [#cache]_
        """
        self._checkscriptname(script)
        for name, _ in self.listscripts(cached=cached):
            if name == script:
                return True
        return False

    def setactive(self, script: str):
        """Mark `script` as the active script."""
        self._checkscriptname(script)
        self._scripts = None
        self.execute('SETACTIVE', script)
        self.logger.info('activated %s', script)

    def unauthenticate(self):
        """Unauthenticate."""
        assert self.capabilities    # nosec B101
        if not self.capabilities.unauthenticate:
            raise CapabilityError('UNAUTHENTICATE: not supported')
        self.execute('UNAUTHENTICATE')
        self.login = ''
        self.owner = ''
        self.logger.info('un-authenticated')

    def unsetactive(self):
        """Deactivate the active script."""
        self._scripts = None
        self.execute('SETACTIVE', '')
        self.logger.info('deactivated active script')

    backup: Backup = Backup.NONE
    """Default backup policy."""

    @classmethod
    def _checkscriptname(cls, script):
        """Check whether `script` is a valid script name.

        Raises:
            ValueError: `script` is *not* a valid script name.
        """
        if not cls.isscriptname(script):
            sanitised = escapectrl(script)
            raise ValueError(f'{sanitised}: bad name')

    _scripts: Optional[list[tuple[str, bool]]] = None
    """Script name/active pairs returned by the last :meth:`listscripts`."""

    _scriptnameregex = re.compile('[^\u0000-\u001f\u0080-\u009f\u2028\u2029]+')
    """Matches script names.

    Must be used with :meth:`re.fullmatch`.

    .. seealso::
        * :RFC:`5198` (sec. 2)
        * :RFC:`5804` (sec. 1.6)
    """


@dataclass
class Capabilities():
    """Server capabilities."""

    @classmethod
    def fromdata(cls,
                 lines: Iterable[Sequence[Word]],
                 strict: bool = False) -> 'Capabilities':
        """Create a :class:`Capabilities` object from lines."""
        obj = cls()
        for words in lines:
            try:
                key = words[0].casefold()   # type: ignore[union-attr]
            except IndexError as err:
                raise ProtocolError('expected word') from err
            except AttributeError as err:
                raise ProtocolError('expected string') from err
            try:
                value, = words[1:]
            except ValueError:
                value = ''
            if not isinstance(value, str):
                raise ProtocolError('expected string')
            if key in ('implementation', 'language', 'owner', 'version'):
                setattr(obj, key, value)
            elif key in ('notify', 'sasl', 'sieve'):
                setattr(obj, key, value.casefold().split())
            elif key in ('starttls', 'unauthenticate'):
                setattr(obj, key, True)
            elif key == 'maxredirects':
                setattr(obj, key, int(value))
            elif strict:
                raise ValueError(f'{key}: unknown capability')
            else:
                obj.notunderstood[key] = value
        return obj

    implementation: Optional[str] = None
    """Server application."""

    sieve: tuple[str, ...] = ()
    """Supported Sieve modules."""

    language: Optional[str] = None
    """Language."""

    maxredirects: Optional[int] = None
    """Maximum number of redirect operations permitted in a script."""

    notify: tuple[str, ...] = ()
    """URI schema parts for supported notification methods."""

    owner: str = ''
    """Canonical name of the user authorised as."""

    sasl: tuple[str, ...] = ()
    """Supported authentication methods."""

    starttls: bool = False
    """Is "STARTTLS" available?"""

    unauthenticate: bool = False
    """Is "UNAUTHENTICATE" available?"""

    version: Optional[str] = None
    """ManageSieve protocol version."""

    notunderstood: dict = dataclasses.field(default_factory=dict)
    """Capabilities not understood by SieveManager."""


@dataclass(frozen=True)
class Response():
    """Server response.

    .. seealso::
        * :RFC:`5804` (secs. 1.2, 1.3, 4, 6.4, and passim)
    """

    @classmethod
    def fromdata(cls, words: list[Word]) -> 'Response':
        """Create a :class:`Response` object from an ACAP line.

        Raises:
            ProtocolError: Server violated ManageSieve protocol.
        """
        # pylint: disable=redefined-outer-name
        code = cls.code
        data = cls.data
        message = cls.message
        for word in words:
            if isinstance(word, Atom):
                res = word.casefold()
            elif isinstance(word, list):
                try:
                    code = word[0].casefold()
                except (AttributeError, IndexError) as err:
                    raise ProtocolError('expected atom') from err
                data = tuple(word[1:])
            elif isinstance(word, str):
                message = word
            else:
                raise ProtocolError('expected atom, list, or string')
        return cls(res, code, data, message)

    def __str__(self) -> str:
        """Human-readable message."""
        return self.message if self.message else f'server says {self.response}'

    def matches(self, *codes: str) -> bool:
        """Check if :attr:`code` matches one of the given `codes`.

        For example:

        >>> with open('script.sieve') as script:
        >>>     try:
        >>>         mgr.putscript(script, script.name)
        >>>     except OperationFailedError as err:
        >>>         if err.matches('quota'):
        >>>             print('over quota')

        Print more informative messages:

        >>> with open('script.sieve') as script:
        >>>     try:
        >>>         mgr.putscript(script, script.name)
        >>>     except OperationFailedError as err:
        >>>         if err.matches('quota/maxscripts'):
        >>>             print('too many scripts')
        >>>         elif err.matches('quota/maxsize'):
        >>>             print(f'{script.name} is too large')
        >>>         elif err.matches('quota'):
        >>>             print('over quota')
        """
        for code_ in codes:
            pattern = re.escape(code_) + r'(/|$)'
            if re.match(pattern, self.code, flags=re.IGNORECASE):
                return True
        return False

    def toerror(self) -> 'Error':
        """Convert `Response` into an error."""
        cls = (ConnectionClosedError if self.response == 'bye'
               else OperationFailedError)
        return cls(self.response, self.code, self.data, self.message)

    response: str
    """``'ok'``, ``'no'``, or ``'bye'``.

    =========  ================================
    Response   Meaning
    =========  ================================
    ``'ok'``   Success
    ``'no'``   Failure
    ``'bye'``  Server has closed the connection
    =========  ================================
    """

    code: str = ''
    """Response code.

    ManageSieve response codes are lists of categories, separated
    by slashes ("/"), where each category is the super-category of
    the next category (e.g., ``('quota/maxsize')``).

    See :RFC:`5804` (sec. 1.3) for a list of response codes.

    .. note::
        Servers need *not* return response codes. Do not rely on them.

    .. seealso::
        * :meth:`matches`
    """

    data: tuple[Union[Atom, int, str], ...] = ()
    """Data attached to the response."""

    message: Optional[str] = None
    """Human-readable message.

    .. note::
        :meth:`Response.__str__` returns a stub message
        if the server did not give a human-readable message.
    """


@dataclass(frozen=True)
class SRV():
    """DNS SRV record."""

    priority: int
    weight: int
    host: str
    port: int


@dataclass(frozen=True)
class URL():
    """Sieve URL.

    .. seealso::
        * :RFC:`5804` (sec. 3)
    """

    @classmethod
    def fromstr(cls, url: str):
        """Create a :class:`URL` object from a URL.

        Raises:
            ValueError: Not a Sieve URL or no host.
        """
        if not cls._schemeregex.match(url):
            url = 'sieve://' + url
        parts = urllib.parse.urlsplit(url)
        if parts.query or parts.fragment:
            raise ValueError('not a Sieve URL')
        if not parts.hostname:
            raise ValueError('no hostname')
        try:
            owner, scriptname = parts.path.split('/', maxsplit=2)[1:]
        except ValueError:
            owner, scriptname = parts.path[1:], None
        return cls(
            scheme=parts.scheme,
            username=parts.username,
            password=parts.password,
            hostname=parts.hostname,
            port=parts.port,
            owner=owner,
            scriptname=scriptname
        )

    def __str__(self):
        """Convert into a URL."""
        url = ''
        if self.scheme:
            url += f'{self.scheme}://'
        else:
            url += 'sieve://'
        if self.username:
            url += self.username
            if self.password is not None:
                url += f':{self.password}'
            url += '@'
        if self.hostname:
            url += self.hostname
        else:
            url += 'localhost'
        if self.port:
            url += f':{self.port}'
        if self.owner:
            url += f'/{self.owner}'
        if self.scriptname:
            url += f'/{self.scriptname}'
        return url

    hostname: str
    scheme: Optional[str] = None
    username: Optional[str] = None
    password: Optional[str] = None
    port: Optional[int] = None
    owner: Optional[str] = None
    scriptname: Optional[str] = None

    _schemeregex = re.compile(r'([a-z][a-z0-9\+\-\.]*:)?//')


def certrevoked(cert: 'x509') -> bool:
    """Check if `cert` may have been revoked.

    Raises:
        OCSPError: Status could not be checked.

    .. seealso::
        * :RFC:`5019`
        * :RFC:`6960`
    """
    try:
        ca_issuers, responders = getcainfo(cert)
    except ExtensionNotFound as err:
        raise OCSPError('no authority information in certificate') from err
    for ca_url in ca_issuers:
        try:
            der = httpget(ca_url)
        except HTTPError as err:
            logging.error(err)
            continue
        ca = x509.load_der_x509_certificate(der)
        builder = ocsp.OCSPRequestBuilder()
        # The use of SHA1 is mandated by RFC 5019.
        req = builder.add_certificate(cert, ca, SHA1()).build()  # nosec B303
        # pylint: disable=redefined-outer-name
        path = b64encode(req.public_bytes(Encoding.DER)).decode('ascii')
        for responder in responders:
            url = urllib.parse.urljoin(responder, path)
            try:
                res = ocsp.load_der_ocsp_response(httpget(url))
            except HTTPError as err:
                logging.error(err)
                continue
            if res.response_status == OCSPResponseStatus.SUCCESSFUL:
                now = datetime.datetime.now()
                if res.this_update <= now < res.next_update:
                    if res.certificate_status == OCSPCertStatus.REVOKED:
                        return True
                    if res.certificate_status == OCSPCertStatus.GOOD:
                        return False
    raise OCSPError('no authoritative OCSP response')


# pylint: disable=redefined-outer-name
def escapectrl(string: str) -> str:
    """Escape control characters."""
    categories = map(unicodedata.category, string)
    # pylint: disable=consider-using-f-string
    escaped = [r'\u%04x' % ord(char) if cat.startswith('C') else char
               for char, cat in zip(string, categories)]
    return ''.join(escaped)


def getcainfo(cert: 'x509') -> tuple[list[str], list[str]]:
    """Get authority information from `cert`.

    Returns:
        CA issuer URLs and OCSP responder base URLs.

    Raises:
        ExtensionNotFound: `cert` does not contain authority information.
    """
    exts = cert.extensions.get_extension_for_class(AuthorityInformationAccess)
    ca_issuers = []
    ocsp = []
    for field in exts.value:
        oid = field.access_method
        if oid == AuthorityInformationAccessOID.CA_ISSUERS:
            ca_issuers.append(field.access_location.value)
        elif oid == AuthorityInformationAccessOID.OCSP:
            ocsp.append(field.access_location.value)
    return ca_issuers, ocsp


def getbackupname(fname: str,
                  getfiles: Callable[[], Iterable[str]] = lambda: (),
                  policy: Backup = Backup.SIMPLE) -> Optional[str]:
    """Derive an Emacs-like filename for a backup.

    Arguments:
        fname: Name of the file to backup.
        getfiles: Function that returns a list of existing files.
        policy: Backup policy.

    .. seealso::
        * `Emacs manual`_ (chap. 19.3.2)
    """
    if policy == Backup.NONE:
        return None
    if policy == Backup.SIMPLE:
        return fname + '~'
    if policy == Backup.NUMBERED:
        regex = re.compile(re.escape(fname) + r'\.~(\d+)~')
        last = 0
        for file in getfiles():
            if match := regex.fullmatch(file):
                if (num := int(match.group(1))) > last:
                    last = num
        rev = last + 1
        return fname + f'.~{rev}~'
    raise ValueError(f'{policy}: no such backup policy')


def getfilesize(file: io.IOBase) -> int:
    """Get the size of file-like object relative to the current position."""
    try:
        pos = file.tell()
    except io.UnsupportedOperation:
        pos = 0
    try:
        size = os.fstat(file.fileno()).st_size - pos
    except io.UnsupportedOperation:
        size = file.seek(0, SEEK_END) - pos
        file.seek(pos, SEEK_SET)
    return size


def httpget(url: str) -> bytes:
    """Download a file from `url`."""
    while True:
        if not url.startswith('http://'):
            raise HTTPError(f'{url}: not an HTTP URL')
        # pylint: disable=consider-using-with
        res = urllib.request.urlopen(url)   # nosec B310
        if res.status == 200:
            return res.read()
        if res.status in (301, 302, 303, 307, 308):
            if url := res.getheader('Location'):
                continue
        raise HTTPError(f'GET {url}: {res.reason}')
    # NOTREACHED


def isipaddress(addr: str) -> bool:
    """Check whether `addr` is an IP address."""
    try:
        ipaddress.ip_address(addr)
    except ValueError:
        return False
    return True


def randomise(elems: Sequence[T], weights: Iterable[int]) -> list[T]:
    """Randomise the order of `elems`."""
    nelems = len(elems)
    weights = list(weights)
    indices = list(range(len(elems)))
    randomised = []
    for _ in range(nelems):
        i, = random.choices(indices, weights, k=1)     # nosec B311
        randomised.append(elems[i])
        weights[i] = 0
    return randomised


def resolvesrv(host: str) -> Iterator[SRV]:
    """Resolve an SRV record.

    Arguments:
        host: Hostname (e.g., :file:`_sieve._tcp.imap.foo.example`)

    Returns:
        An iterator over `SRV` records sorted by their priority
        and randomised according to their weight.

    .. note::
        Requires dnspython_.

    .. _dnspython: https://www.dnspython.org/
    """
    answer = dns.resolver.resolve(host, 'SRV')
    hosts: dict[str, list] = defaultdict(list)
    byprio: dict[int, list[SRV]] = defaultdict(list)
    for rec in answer.response.additional:
        name = rec.name.to_text()
        addrs = [i.address for i in rec.items if i.rdtype == dns.rdatatype.A]
        hosts[name].extend(addrs)
    for rec in answer:
        name = rec.target.to_text()
        if addrs := hosts.get(name):    # type: ignore[assignment]
            srvs = [SRV(rec.priority, rec.weight, a, rec.port) for a in addrs]
            byprio[rec.priority].extend(srvs)
        else:
            srv = SRV(rec.priority, rec.weight, name.rstrip('.'), rec.port)
            byprio[rec.priority].append(srv)
    for prio in sorted(byprio.keys()):
        srvs = byprio[prio]
        weights = [s.weight for s in srvs]
        if sum(weights) > 0:
            yield from randomise(srvs, weights)
        else:
            yield from srvs


#
# Authentication
#

class AbstractSASLChannel(ABC):
    """Protocol for connections to perform SASL authentication over."""

    @abstractmethod
    def abort(self):
        """Abort the authentication process."""

    @abstractmethod
    def begin(self, name: str, data: Optional[bytes] = None):
        """Begin the authentication process.

        Arguments:
            name: Name of the SASL mechanism used.
            data: Optional client-first message.
        """

    @abstractmethod
    def end(self):
        """Conclude the authentication process.

        Raises:
            OperationError: Authentication failed.
        """

    @abstractmethod
    def send(self, data: bytes):
        """Encode and send an SASL message."""

    @abstractmethod
    def receive(self) -> bytes:
        """Receive and decode an SASL message."""

    sock: Union[socket.SocketType, ssl.SSLSocket]
    """Underlying socket."""


class SieveSASLChannel(AbstractSASLChannel):
    """Adapter to send SASL messages over a :class:`SieveConnection`."""

    def __init__(self, connection: 'SieveConnection',
                 log: bool = False):
        """Initialise the adapter.

        Arguments:
            connection: ManageSieve connection.
            log: Log the authentication exchange?
        """
        self.conn = connection
        self.log = log

    def abort(self):
        """Abort the authentication process."""
        self.send(b'*')
        self.end()

    def begin(self, name: str, data: Optional[bytes] = None):
        """Begin the authentication process.

        Arguments:
            name: Name of the SASL mechanism used.
            data: Optional client-first message.
        """
        assert self.conn  # nosec B101
        args: list[Union[str, bytes]] = [Atom('AUTHENTICATE'), name.upper()]
        if data is not None:
            args.append(b64encode(data))
        self.conn.sendline(*args, log=self.log)

    def end(self):
        """Conclude the authentication process.

        Raises:
            OperationFailedError: Authentication failed.
            ProtocolError: Server violated SASL protocol.
        """
        assert self.conn  # nosec B101
        res = Response.fromdata(self.conn.receiveline(log=self.log))
        if res.response != 'ok':
            raise res.toerror()

    def send(self, data: bytes):
        """Encode and send an SASL message."""
        assert self.conn  # nosec B101
        conn = self.conn
        conn.sendline(b64encode(data), log=self.log)

    def receive(self) -> bytes:
        """Receive and decode an SASL message."""
        assert self.conn  # nosec B101
        words = self.conn.receiveline(log=self.log)
        if isinstance(words[0], Atom):
            res = Response.fromdata(words)
            raise res.toerror()
        try:
            word, = words
        except ValueError as err:
            raise ProtocolError('unexpected data') from err
        if isinstance(word, Atom) or not isinstance(word, str):
            raise ProtocolError('expected string')
        return b64decode(word)

    @property
    def sock(self) -> Union[socket.SocketType, ssl.SSLSocket]:
        """Underlying socket."""
        assert self.conn        # nosec B101
        assert self.conn.sock   # nosec B101
        return self.conn.sock

    @sock.setter
    def sock(self, sock: Union[socket.SocketType, ssl.SSLSocket]):
        assert self.conn        # nosec B101
        self.conn.sock = sock

    conn: Optional[SieveConnection] = None
    """Underlying connection."""


class AbstractAuth(ABC):
    """Protocol for SASL authentication processes."""

    @abstractmethod
    def __init__(self,
                 channel: AbstractSASLChannel,
                 authcid: str,
                 authzid: str = '',
                 prepare: SASLPrep = SASLPrep.ALL):
        """Prepare authentication.

        Arguments:
            authcid: Authentication ID (user to login as).
            authzid: Authorisation ID (use whose rights to acquire).
            prepare: Which credentials to string prepare.

        Raises:
            ValueError: Bad characters in username.

        .. seealso::
            * :RFC:`4013`
            * :RFC:`4616`
        """

    @abstractmethod
    def __call__(self) -> Optional[Any]:
        """Authenticate.

        Returns:
            Data returned by the server, if any.
        """

    name: ClassVar[str]
    """Mechanism name."""

    authcid: str
    """Authentication ID (user to login as)."""

    authzid: str = ''
    """Authorisation ID (user whose scripts to manage)."""


class BaseAuth(AbstractAuth, ABC):
    """Base class for authentication mechanisms.

    .. note::
        :class:`BaseAuth` handles SASL compression transparently.
        Most subclasses should only override :meth:`__init__` and
        :meth:`exchange` and only use :meth:`send` and :meth:`receive`.
    """

    @staticmethod
    # pylint: disable=redefined-outer-name
    def prepare(string: str) -> str:
        """Prepare a string according to the "SASLprep" profile.

        Raises:
            ValueError: String contains bad characters.

        .. seealso::
            * :RFC:`4013`
        """
        prep = ''
        normalised = unicodedata.ucd_3_2_0.normalize('NFKC', string)
        for i, char in enumerate(normalised, start=1):
            if stringprep.in_table_b1(char):
                pass
            elif stringprep.in_table_c12(char):
                prep += ' '
            elif stringprep.in_table_c21_c22(char):
                raise ValueError(f'{string}:{i}: control character')
            elif stringprep.in_table_c3(char):
                raise ValueError(f'{string}:{i}: private use character')
            elif stringprep.in_table_c4(char):
                raise ValueError(f'{string}:{i}: non-character code point')
            elif stringprep.in_table_c5(char):
                raise ValueError(f'{string}:{i}: surrogate code point')
            elif stringprep.in_table_c6(char):
                raise ValueError(f'{string}:{i}: not plain text')
            elif stringprep.in_table_c7(char):
                raise ValueError(f'{string}:{i}: not canonical')
            elif stringprep.in_table_c8(char):
                raise ValueError(f'{string}:{i}: changes display/deprecated')
            elif stringprep.in_table_c9(char):
                raise ValueError(f'{string}:{i}: tagging character')
            else:
                prep += char
        return prep

    @classmethod
    def subclasses(cls, sort: bool = True) -> Iterable[type['BaseAuth']]:
        """Get non-abstract and non-obsolete subclasses.

        Arguments:
            sort: Sort classes by :attr:`order`?
        """
        classes: set[type['BaseAuth']] = set()
        for subcls in cls.__subclasses__():
            if not subcls.__abstractmethods__ and not subcls.obsolete:
                classes.add(subcls)
            classes.update(subcls.subclasses(sort=False))
        if sort:
            return list(sorted(classes, key=lambda c: c.order))
        return classes

    def __init__(self,
                 channel: AbstractSASLChannel,
                 authcid: str,
                 authzid: str = '',
                 prepare: SASLPrep = SASLPrep.ALL):
        """Prepare authentication.

        Arguments:
            authcid: Authentication ID (user to login as).
            authzid: Authorisation ID (use whose rights to acquire).
            prepare: Which credentials to :meth:`prepare`.

        Raises:
            ValueError: Bad characters in username.
        """
        prepare &= SASLPrep.USERNAMES   # type: ignore[assignment]
        self.channel = channel
        self.authcid = self.prepare(authcid) if prepare else authcid
        self.authzid = self.prepare(authzid) if prepare else authzid

    # pylint: disable=useless-return
    def __call__(self) -> Optional[Capabilities]:
        """Authenticate.

        Calls :meth:`exchange` and :meth:`end`.
        """
        self.exchange()
        if self.phase == AuthPhase.RECEIVED:
            self.send(b'')
        self.end()
        return None

    def abort(self):
        """Abort authentication."""
        self.channel.abort()
        self.phase = AuthPhase.DONE

    def begin(self, data: Optional[bytes] = None):
        """Start the authentication process if needed.

        Arguments:
            data: Optional client-first message.
        """
        if self.phase == AuthPhase.PREAUTH:
            self.channel.begin(self.name, data)
            self.phase = AuthPhase.SENT
        else:
            raise ProtocolError('unexpected authentication phase')

    def end(self):
        """Conclude the authentication process."""
        if self.phase == AuthPhase.SENT:
            self.channel.end()
        else:
            raise ProtocolError('unexpected authentication phase')

    @abstractmethod
    def exchange(self):
        """Exchange SASL messages."""

    def send(self, data: bytes):
        """Encode and send SASL data.

        .. note::
            Starts the authentication process, if needed.
        """
        if self.phase == AuthPhase.PREAUTH:
            self.begin(data)
        elif self.phase == AuthPhase.RECEIVED:
            self.channel.send(data)
            self.phase = AuthPhase.SENT
        else:
            raise ProtocolError('unexpected authentication phase')

    def receive(self) -> bytes:
        """Receive and decode SASL data.

        .. note::
            Starts the authentication process, if needed.
        """
        if self.phase == AuthPhase.PREAUTH:
            self.begin()
        if self.phase == AuthPhase.SENT:
            try:
                data = self.channel.receive()
                self.phase = AuthPhase.RECEIVED
            except OperationFailedError as err:
                if err.response != 'ok':
                    raise err
                if not err.matches('sasl'):
                    # pylint: disable=raise-missing-from
                    raise ProtocolError('expected data')
                try:
                    word, = err.data
                except ValueError as valuerr:
                    raise ProtocolError('unexpected data') from valuerr
                if isinstance(word, Atom) or not isinstance(word, str):
                    # pylint: disable=raise-missing-from
                    raise ProtocolError('expected string')
                data = b64decode(word)
                self.phase = AuthPhase.DONE
            return data
        raise ProtocolError('unexpected authentication phase')

    @property
    def sock(self) -> Union[socket.SocketType, ssl.SSLSocket]:
        """Underlying socket."""
        assert self.channel   # nosec B101
        return self.channel.sock

    @sock.setter
    def sock(self, sock: Union[socket.SocketType, ssl.SSLSocket]):
        assert self.channel   # nosec B101
        self.channel.sock = sock

    channel: AbstractSASLChannel
    """Underlying SASL channel."""

    obsolete: bool = False
    """Is this mechanism obsolete?"""

    order: int = 0
    """Mechanism precedence."""

    phase: AuthPhase = AuthPhase.PREAUTH
    """Current authentication phase."""

    type: ClassVar[AuthType] = AuthType.OTHER
    """Authentication type."""


class BasePassAuth(BaseAuth, ABC):
    """Base class for password-based authentication mechanisms."""

    def __init__(self,
                 connection: AbstractSASLChannel,
                 authcid: str,
                 password: str,
                 authzid: str = '',
                 prepare: SASLPrep = SASLPrep.ALL):
        """Prepare authentication.

        Arguments:
            authcid: Authentication ID (user to login as).
            authzid: Authorisation ID (use whose rights to acquire).
            password: Password.
            prepare: Which credentials to :meth:`BaseAuth.prepare`.

        Raises:
            ValueError: Bad characters in username or password.
        """
        super().__init__(connection, authcid, authzid, prepare)
        prepare &= SASLPrep.PASSWORDS   # type: ignore[assignment]
        self.password = self.prepare(password) if prepare else password

    password: str
    """Password."""

    type = AuthType.PASSWORD


class BaseScramAuth(BasePassAuth, ABC):
    """Base class for SCRAM authentication mechanisms.

    .. seealso::
        * :RFC:`5802`
        * :RFC:`7677`
        * https://datatracker.ietf.org/doc/html/draft-melnikov-scram-bis
        * https://csb.stevekerrison.com/post/2022-01-channel-binding
        * https://csb.stevekerrison.com/post/2022-05-scram-detail
        * https://blog.josefsson.org/2021/06/08/whats-wrong-with-scram
        * https://github.com/horazont/aiosasl
    """

    # pylint: disable=too-many-locals
    def exchange(self):
        """Exchange SASL messages.

        Raises:
            VerificationError: Server could not be verified.
        """
        def todict(msg: bytes) -> dict[bytes, bytes]:
            return dict([a.split(b'=', maxsplit=1) for a in msg.split(b',')])

        # Parameters
        authcid = self.authcid.encode('utf8')
        authzid = self.authzid.encode('utf8')
        password = self.password.encode('utf8')
        chan_bind_type = self.cbtype.encode('utf8')
        chan_bind_data = self.cbdata
        c_nonce_len = self.noncelen
        digest = self.digest

        # Send client-first message
        chan_bind_attr = b'p=%s' % chan_bind_type if chan_bind_type else b'n'
        c_first_prefix = chan_bind_attr + b',' + authzid
        c_nonce = b64encode(secrets.token_bytes(c_nonce_len))
        c_first_bare = b'n=%s,r=%s' % (authcid, c_nonce)
        c_first = c_first_prefix + b',' + c_first_bare
        self.send(c_first)

        # Receive server-first message
        try:
            s_first = self.receive()
        except OperationError as err:
            raise CapabilityError(f'{self.name}: rejected by server') from err
        s_first_dict = todict(s_first)
        iters = int(s_first_dict[b'i'])
        s_nonce = s_first_dict[b'r']
        salt = b64decode(s_first_dict[b's'])

        # Send client-final message
        salted_pwd = hashlib.pbkdf2_hmac(digest, password, salt, iters)
        c_key = hmac.digest(salted_pwd, b'Client Key', digest)
        stored_key = hashlib.new(digest, c_key).digest()
        chan_bind = b64encode(c_first_prefix + b',' + chan_bind_data)
        c_final_prefix = b'c=%s,r=%s' % (chan_bind, s_nonce)
        auth_message = b','.join((c_first_bare, s_first, c_final_prefix))
        c_signature = hmac.digest(stored_key, auth_message, digest)
        c_proof = b64encode(bytes(a ^ b for a, b in zip(c_key, c_signature)))
        c_final = c_final_prefix + b',p=%s' % c_proof
        self.send(c_final)

        # Receive server-final message
        s_final = self.receive()
        s_key = hmac.digest(salted_pwd, b'Server Key', digest)
        s_signature = hmac.digest(s_key, auth_message, digest)
        s_final_dict = todict(s_final)
        if s_signature != b64decode(s_final_dict[b'v']):
            raise VerificationError('server verification failed')

    @property
    @abstractmethod
    def digest(self) -> str:
        """Digest name as used by :mod:`hashlib` and :mod:`hmac`."""

    cbtype: str = ''
    """TLS channel-binding type."""

    cbdata: bytes = b''
    """TLS channel-binding data."""

    noncelen: int = 18
    """Client nonce length in bytes."""


class BaseScramPlusAuth(BaseScramAuth, ABC):
    """Base class for SCRAM mechanisms with channel binding.

    .. seealso::
        * :class:`BaseScramAuth`
    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        if not isinstance(self.sock, ssl.SSLSocket):
            raise CapabilityError('non-TLS channels cannot be bound')
        for cbtype in ('tls-exporter', 'tls-unique', 'tls-server-endpoint'):
            if cbtype in ssl.CHANNEL_BINDING_TYPES:
                if cbdata := self.sock.get_channel_binding(cbtype):
                    self.cbtype = cbtype
                    self.cbdata = cbdata
                    break
        else:
            raise CapabilityError('channel-binding not supported')


class AuthorisationUnsupportedTrait():
    """Trait for SASL mechanisms that do not support authorisation."""

    # pylint: disable=no-member
    def __init__(self, *args, **kwargs):
        """Initialise the SASL mechanism.

        Raises:
            ProxyAuthError: :attr:`authzid` is set."""
        assert isinstance(self, BaseAuth)   # nosec B101
        super().__init__(*args, **kwargs)
        if self.authzid:
            message = f'{self.name}: authorisation not supported'
            raise ProxyAuthError(message)


class CramMD5Auth(AuthorisationUnsupportedTrait, BasePassAuth):
    """"CRAM-MD5" authentication.

    .. seealso::
        * :RFC:`2195` (sec. 2)
    """

    def exchange(self):
        challenge = self.receive()
        password = self.password.encode('utf8')
        digest = hmac.new(password, challenge, hashlib.md5)
        data = ' '.join((self.authcid, digest.hexdigest()))
        self.send(data.encode('utf8'))

    obsolete = True
    name = 'CRAM-MD5'


class ExternalAuth(BaseAuth):
    """"EXTERNAL" authentication.

    .. seealso::
        * :RFC:`4422` (App. A)
    """

    def __call__(self):
        """Authenticate."""
        args = (self.authzid.encode('utf8'),) if self.authzid else ()
        self.begin(*args)
        self.receive()
        self.send(b'')
        self.end()

    def exchange(self):
        """No-op."""

    name = 'EXTERNAL'
    type = AuthType.EXTERNAL


class LoginAuth(AuthorisationUnsupportedTrait, BasePassAuth):
    """"LOGIN" authentication.

    .. seealso::
        * https://datatracker.ietf.org/doc/draft-murchison-sasl-login
    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        if {self.password} & {'\0', '\f', '\n'}:
            raise ValueError('bad characters in password')

    def exchange(self):
        self.receive()
        self.send(self.authcid.encode('utf8'))
        self.receive()
        self.send(self.password.encode('utf8'))

    obsolete = True
    name = 'LOGIN'


class PlainAuth(BasePassAuth):
    """"PLAIN" authentication.

    .. seealso::
        * :RFC:`4616`
    """

    def exchange(self):
        data = '\0'.join((self.authzid, self.authcid, self.password))
        self.send(data.encode('utf8'))

    name = 'PLAIN'


class ScramSHA1Auth(BaseScramAuth):
    """SCRAM-SHA-1 authentication."""
    digest = 'sha1'
    name = 'SCRAM-SHA-1'
    order = -10


# pylint: disable=too-many-ancestors
class ScramSHA1PlusAuth(BaseScramPlusAuth, ScramSHA1Auth):
    """SCRAM-SHA-1-PLUS authentication."""
    name = 'SCRAM-SHA-1-PLUS'
    order = -1000


class ScramSHA224Auth(BaseScramAuth):
    """SCRAM-SHA-224 authentication."""
    digest = 'sha224'
    name = 'SCRAM-SHA-224'
    order = -20


# pylint: disable=too-many-ancestors
class ScramSHA224PlusAuth(BaseScramPlusAuth, ScramSHA224Auth):
    """SCRAM-SHA-224-PLUS authentication."""
    name = 'SCRAM-SHA-224-PLUS'
    order = -2000


class ScramSHA256Auth(BaseScramAuth):
    """SCRAM-SHA-256 authentication."""
    digest = 'sha256'
    name = 'SCRAM-SHA-256'
    order = -30


# pylint: disable=too-many-ancestors
class ScramSHA256PlusAuth(BaseScramPlusAuth, ScramSHA256Auth):
    """SCRAM-SHA-256-PLUS authentication."""
    name = 'SCRAM-SHA-256-PLUS'
    order = -3000


class ScramSHA384Auth(BaseScramAuth):
    """SCRAM-SHA-384 authentication."""
    digest = 'sha384'
    name = 'SCRAM-SHA-384'
    order = -40


# pylint: disable=too-many-ancestors
class ScramSHA384PlusAuth(BaseScramPlusAuth, ScramSHA384Auth):
    """SCRAM-SHA-384-PLUS authentication."""
    name = 'SCRAM-SHA-384-PLUS'
    order = -4000


class ScramSHA512Auth(BaseScramAuth):
    """SCRAM-SHA-512 authentication."""
    digest = 'sha512'
    name = 'SCRAM-SHA-512'
    order = -50


# pylint: disable=too-many-ancestors
class ScramSHA512PlusAuth(BaseScramPlusAuth, ScramSHA512Auth):
    """SCRAM-SHA-512-PLUS authentication."""
    name = 'SCRAM-SHA-512-PLUS'
    order = -5000


# pylint: disable=invalid-name
class ScramSHA3_512Auth(BaseScramAuth):
    """SCRAM-SHA-512 authentication."""
    digest = 'sha3_512'
    name = 'SCRAM-SHA3-512'
    order = -60


# pylint: disable=invalid-name,too-many-ancestors
class ScramSHA3_512PlusAuth(BaseScramPlusAuth, ScramSHA3_512Auth):
    """SCRAM-SHA-512-PLUS authentication."""
    name = 'SCRAM-SHA3-512-PLUS'
    order = -6000


#
# SieveManager Shell
#

class BaseShell():
    """Base class for shells."""

    def __init__(self):
        """Initialise a :class:`BaseShell` object."""
        self._completions: list = []
        self.commands = tuple(a[3:] for a in dir(self) if a.startswith('do_'))

    def complete(self, text, n) -> Optional[str]:
        """Return the `n`-th completion for `text`.

        Tab-completion for commands is built-in, but tab-completion for
        arguments must be enabled by provided a :samp:`complete_{command}`
        method for each `command`.

        .. seealso::
            :func:`readline.set_completer`
        """
        if n == 0:
            logger = logging.getLogger(__name__)
            loglevel = logger.level
            logger.setLevel(logging.ERROR)
            try:
                begin = readline.get_begidx()
                buffer = readline.get_line_buffer()
                args = self.expand(buffer[:begin])
                argidx = len(args)
                if argidx == 0:
                    words = self.commands
                else:
                    command = 'help' if args[0] == '?' else args[0]
                    try:
                        complete = getattr(self, f'complete_{command}')
                    except AttributeError:
                        return None
                    words = complete(argidx, text)
                self._completions = sorted(fnmatch.filter(words, text + '*'))
            finally:
                logger.setLevel(loglevel)
        try:
            return self._completions[n]
        except (AttributeError, IndexError):
            return None

    def confirm(self, prompt: str, multi: bool = False) -> ConfirmEnum:
        """Prompt for confirmation.

        Arguments:
            prompt: Prompt.
            multi: Permit choices "all" and "none"?
        """
        if not os.isatty(sys.stdin.fileno()):
            return ConfirmEnum.YES
        while True:
            answer = input(prompt + ' [no] ').casefold()
            if answer in ('y', 'yes'):
                return ConfirmEnum.YES
            if answer in ('', 'n', 'no'):
                return ConfirmEnum.NO
            if multi:
                if answer == 'all':
                    return ConfirmEnum.ALL
                if answer == 'none':
                    return ConfirmEnum.NONE
            if multi:
                print('Enter "yes", "no", "all", or "none"')
            else:
                print('Enter "yes" or "no"')
        # NOTREACHED

    def enter(self):
        """Enter the shell.

        The shell is exited if the end-of-filer marker is reached
        or if a command raises :exc:`StopIteration`.

        Commands must be provided by :samp:`do_{command}` methods.
        "?" is used as alias for "help", "!" as alias for "shell".

        .. note::
            Entering the shell binds :kbd:`tab` to :func:`rl_complete`
            if :mod:`readline` is implemented by `libedit`.
        """
        logger = self.logger
        oldcompleter = readline.get_completer()
        readline.set_auto_history(True)
        readline.set_completer(self.complete)
        readline.parse_and_bind('tab: complete')
        if 'libedit' in readline.__doc__:
            readline.parse_and_bind('python:bind ^I rl_complete')
        logger.info('enter "? [command]" for help and "exit" to exit')
        try:
            while True:
                readline.set_auto_history(True)
                readline.set_completer(self.complete)
                try:
                    line = input(self.getprompt()).strip()
                except EOFError:
                    break
                readline.set_auto_history(False)
                # Setting to None doesn't work with libedit.
                readline.set_completer(lambda *_: None)
                try:
                    self.run(line)
                except (ConnectionError, ProtocolError) as err:
                    raise err
                except subprocess.CalledProcessError as err:
                    logger.error('%s: exited with status %d',
                                 err.cmd[0], err.returncode)
                except (FileNotFoundError, FileExistsError) as err:
                    logger.error('%s: %s', err.filename,
                                 os.strerror(err.errno))
                except (GetoptError, OSError, ValueError, Error) as err:
                    for line in str(err).splitlines():
                        logger.error('%s', line)
                except StopIteration:
                    break
        finally:
            readline.set_auto_history(True)
            readline.set_completer(oldcompleter)

    # pylint: disable=too-many-branches, too-many-statements
    def expand(self, line: str, glob: bool = False) -> list[str]:
        """Split `line` into words as a POSIX-compliant shell would.

        Wildcards are expanded to the words provided by tab-completion.

        Arguments:
            line: Line.
            glob: Expand :func:`fnmatch.fnmatch` patterns?
        """
        buffer: str = ''
        escape: bool = False
        expand: bool = False
        quotes: list[str] = []
        tokens: list[tuple[bool, str]] = []
        for i, char in enumerate(line):
            if escape:
                if glob and char in '*?[]':
                    buffer += '\\'
                buffer += char
                escape = False
            elif quotes:
                if char in '\'"':
                    if char == quotes[-1]:
                        del quotes[-1]
                    else:
                        quotes.append(char)
                    if quotes:
                        buffer += char
                elif char == '\\':
                    escape = True
                else:
                    if glob and char in '*?[]':
                        buffer += '\\'
                    buffer += char
            elif char == '\\':
                escape = True
            elif char in '\'"':
                quotes.append(char)
            elif char.isspace():
                if i > 0 and not line[i - 1].isspace():
                    tokens.append((expand, buffer))
                    buffer = ''
                    expand = False
            else:
                if glob and char in '*?[]':
                    expand = True
                buffer += char
        if buffer or tokens:
            tokens.append((expand, buffer))
        expanded = []
        if glob:
            try:
                getfiles = getattr(self, f'complete_{tokens[0][1]}')
            except (IndexError, AttributeError):
                getfiles = None
        for i, (expand, token) in enumerate(tokens):
            if expand and getfiles:
                matches = sorted(fname for fname in getfiles(i, '')
                                 if (not fname.startswith('.')
                                     and fnmatch.fnmatchcase(fname, token)))
                if not matches:
                    raise ValueError(f'{token}: no matches')
                expanded.extend(matches)
            else:
                if glob:
                    token = re.sub(r'\\([*?\[\]])', r'\1', token)
                expanded.append(token)
        return expanded

    def getprompt(self) -> str:
        """Get a shell prompt."""
        return '> '

    def run(self, line: str):
        """Run `line` as a :class:`BaseShell` command.

        Raises:
            UsageError: Command not found or arguments wrong.
        """
        if line.startswith('?'):
            name = 'help'
            args = self.expand(line[1:])
        elif line.startswith('!'):
            name = 'sh'
            args = self.expand(line[1:])
        else:
            try:
                name, *args = self.expand(line, glob=True)
            except ValueError:
                return
        try:
            command = getattr(self, f'do_{name}')
        except AttributeError as err:
            raise UsageError(f'{name}: no such command') from err
        try:
            inspect.signature(command).bind(*args)
        except TypeError as err:
            usage, _ = command.__doc__.split(' - ', maxsplit=1)
            raise UsageError(f'usage: {usage}') from err
        command(*args)

    commands: tuple[str, ...]
    """Shell commands."""

    logger: logging.Logger = logging.getLogger(__name__)
    """Logger.

    :class:`SieveShell` logs messages of the following priorities:

    * :const:`logging.ERROR`
    * :const:`logging.INFO`

    .. seealso::
        * :attr:`SieveConnection.logger`
    """


class SieveShell(BaseShell):
    """Shell around a `SieveManager` object."""

    def __init__(self, manager: 'SieveManager',
                 confirm: ShellCmd = ShellCmd.ALL):
        """Initialise a :class:`SieveShell` object.

        Arguments:
            manager: Connection to a ManageSieve server.
            confirm: Shell commands that require confirmation if
                they overwrite or remove files.
        """
        super().__init__()
        self._confirm = confirm
        self._overwrite: Optional[ConfirmEnum] = None
        self.manager = manager

    # Methods
    def getprompt(self):
        ps = super().getprompt()
        mgr = self.manager
        if mgr and (url := mgr.geturl()) is not None:
            ps = ('' if mgr.tls else '(insecure) ') + str(url) + ps
        return ps

    # Shell commands
    def do_activate(self, script: str):
        """activate script - mark script as active"""
        self.manager.setactive(script)

    def do_caps(self):
        """caps - show server capabilities"""
        if (caps := self.manager.capabilities) is not None:
            indent = ' ' * 4
            for key, value in caps.__dict__.items():
                if value:
                    if key in ('implementation', 'language', 'maxredirects',
                               'owner', 'version'):
                        print(f'{key}: {value}')
                    elif key in ('notify', 'sasl', 'sieve'):
                        print(f'{key}:')
                        for v in value:
                            print(f'{indent}- {v}')
                    elif key in ('starttls', 'unauthenticate'):
                        print(f'{key}: yes')
            for key, value in caps.notunderstood.items():
                if value is not None:
                    print(f'{key}: {value}')

    def do_cat(self, *scripts: str):
        """cat [script ...] - concatenate scripts on standard output"""
        for script in scripts:
            sys.stdout.write(self.manager.getscript(script))

    def do_cd(self, localdir: str = HOMEDIR):
        """cd [localdir] - change local directory"""
        os.chdir(localdir)
        self.logger.info('changed directory to %s', localdir)

    def do_cert(self):
        """cert - show the server's TLS certificate."""
        indent = ' ' * 4
        mgr = self.manager
        if not isinstance(mgr.sock, ssl.SSLSocket):
            raise UsageError('not a secure connection')
        cert = mgr.sock.getpeercert()
        assert cert     # nosec B101
        value: Any
        for key, value in sorted(cert.items()):
            if key in ('OCSP', 'caIssuers', 'crlDistributionPoints'):
                print(f'{key}:')
                for item in sorted(value):
                    print(f'{indent}- {item}')
            elif key in ('issuer', 'subject'):
                print(f'{key}:')
                for pairs in sorted(value):
                    print(f'{indent}- ', end='')
                    for i, (k, v) in enumerate(pairs):
                        if i == 0:
                            print(f'{k}: {v}')
                        else:
                            print(f'{indent}  {k}: {v}')
            elif key == 'subjectAltName':
                print(f'{key}:')
                for k, v in sorted(value):
                    print(f'{indent}- {k}: {v}')
            elif isinstance(value, (int, str)):
                print(f'{key}: {value}')

    def do_cmp(self, script1: str, script2: str):
        """cmp script1 script2 - compare two scripts"""
        self.manager.editscripts(['cmp'], [script1, script2], check=False)

    def do_cp(self, source: str, target: str):
        """cp source target - download source and re-upload it as target."""
        if self.manager.scriptexists(target):
            if self._confirm & ShellCmd.CP:
                if not self.confirm(f'Overwrite {target}?'):
                    raise FileExistsError(EEXIST, os.strerror(EEXIST), target)
        self.manager.copyscript(source, target)

    def do_check(self, localscript: str):
        """check localscript - check whether localscript is valid"""
        with open(localscript, 'rb') as file:
            # checkscript performs a blocking send/sendline,
            # but holding a lock should be okay in an interactive app.
            fcntl.flock(file.fileno(), LOCK_SH | LOCK_NB)
            self.manager.checkscript(file)
        self.logger.info('%s is valid', localscript)

    def do_deactivate(self):
        """deactivate - deactivate the active script"""
        self.manager.unsetactive()

    def do_diff(self, script1: str, script2: str):
        """diff script1 script2 - show differences between scripts"""
        self.manager.editscripts(['diff'], [script1, script2], check=False)

    def do_ed(self, *args: str):
        """ed [-a] script [...] - edit scripts with a line editor"""
        mgr = self.manager
        (opts, scripts) = getopt(list(args), 'a')
        for opt, _ in opts:
            if opt == '-a':
                scripts = [script] if (script := mgr.getactive()) else []
        if scripts:
            editor = os.getenv('EDITOR', 'ed')
            mgr.editscripts([editor], list(scripts))
        else:
            self.logger.error('no scripts given')

    def do_exit(self) -> bool:
        """exit - exit SieveManager"""
        raise StopIteration()

    def do_get(self, source: str, target: Optional[str] = None,
               srcidx: Optional[int] = None):
        """get source [target] - download script"""
        if not target:
            target = path.basename(source.rstrip('/'))
        if not srcidx:
            self._overwrite = None
        flags = O_CREAT | O_EXCL | O_WRONLY | O_TRUNC
        # Try not to make a backup if the source doesn't exist.
        # Writing to a temporary file would create a worse race condition.
        policy = self.manager.backup
        if policy != Backup.NONE:
            if not self.manager.scriptexists(source):
                raise FileNotFoundError(ENOENT, os.strerror(ENOENT), source)
        if path.exists(target):
            if self._confirm & ShellCmd.GET:
                if self._overwrite not in \
                        (ConfirmEnum.NONE, ConfirmEnum.ALL):
                    self._overwrite = \
                        self.confirm(f'Overwrite {target}?', bool(srcidx))
                if not self._overwrite:
                    raise FileExistsError(EEXIST, os.strerror(EEXIST), target)
            def getfiles():
                return readdir(path.dirname(target), path.isfile)
            if backup := getbackupname(target, getfiles, policy):
                shutil.copy(target, backup)
                self.logger.info('backed up %s as %s', target, backup)
            flags &= ~O_EXCL
        fd = os.open(target, flags, mode=0o644)
        # getscript performs a blocking read,
        # but holding a lock should be okay in an interactive app.
        fcntl.flock(fd, LOCK_SH | LOCK_NB)
        with os.fdopen(fd, 'w') as file:
            file.write(self.manager.getscript(source))
        self.logger.info('downloaded %s', target)

    def do_help(self, name: Optional[str] = None):
        """help [command] - list commands/show help for command"""
        if name:
            try:
                print(getattr(self, f'do_{name}').__doc__)
            except AttributeError:
                self.logger.error('%s: unknown command', name)
        else:
            columnise(self.commands)

    def do_ls(self, *args: str):
        """ls [-a] [script ...] - list scripts"""
        active = False
        (opts, fnames) = getopt(list(args), 'a')
        for opt, _ in opts:
            if opt == '-a':
                active = True
        if active:
            if script := self.manager.getactive():
                print(script)
            else:
                self.logger.warning('no active script')
        else:
            scripts = self.manager.listscripts()
            if fnames:
                existing = {fname for fname, _ in scripts}
                for name in set(fnames) - existing:
                    raise FileNotFoundError(ENOENT, os.strerror(ENOENT), name)
                scripts = [s for s in scripts if s[0] in fnames]
            words = [fname + ('*' if active else '')
                     for fname, active in scripts]
            columnise(sorted(words))

    def do_mget(self, *scripts: str):
        """mget script [...] - download multiple scripts"""
        for i, script in enumerate(scripts, start=1):
            self.do_get(script, srcidx=i)

    def do_more(self, *args: str):
        """more [-a] script [...] - display scripts page-by-page."""
        mgr = self.manager
        (opts, scripts) = getopt(list(args), 'a')
        for opt, _ in opts:
            if opt == '-a':
                scripts = [script] if (script := mgr.getactive()) else []
        mgr.editscripts([os.getenv('PAGER', 'more')], list(scripts))

    def do_mput(self, *scripts: str):
        """mput script [...] - upload multiple scripts"""
        for i, script in enumerate(scripts, start=1):
            self.do_put(script, srcidx=i)

    def do_mv(self, source: str, target: str):
        """mv source target - rename source to target"""
        mgr = self.manager
        if mgr.scriptexists(target):
            if self._confirm & ShellCmd.MV:
                if not self.confirm(f'Overwrite {target}?'):
                    raise FileExistsError(EEXIST, os.strerror(EEXIST), target)
            mgr.backupscript(target)
            mgr.deletescript(target)
        mgr.renamescript(source, target, emulate=True)

    def do_put(self, source: str, target: Optional[str] = None,
               srcidx: Optional[int] = None):
        """put source [target] - upload script"""
        mgr = self.manager
        if not srcidx:
            self._overwrite = None
        if not target:
            target = path.basename(source.rstrip('/'))
        if mgr.scriptexists(target):
            if self._confirm & ShellCmd.PUT:
                if self._overwrite not in (ConfirmEnum.ALL, ConfirmEnum.NONE):
                    self._overwrite = \
                        self.confirm(f'Overwrite {target}?', bool(srcidx))
                if not self._overwrite:
                    raise FileExistsError(EEXIST, os.strerror(EEXIST), target)
        with open(source, 'rb') as file:
            # checkscript performs a blocking send/sendline,
            # but holding a lock should be okay in an interactive app.
            fcntl.flock(file.fileno(), LOCK_SH | LOCK_NB)
            mgr.putscript(file, target)

    def do_python(self):
        """python - enter Python read-evaluate-print loop"""
        try:
            readline.clear_history()
        except AttributeError:
            pass
        try:
            banner = (f'Python {sys.version}\n'
                       'enter "dir()" to list objects, '
                       '"help()" for help, and '
                       '"exit()" to exit')
            code.interact(local=InteractiveProxy(self.manager),
                          banner=banner, exitmsg='')
        except SystemExit:
            pass

    def do_rm(self, *scripts: str):
        """rm [script ...] - remove script"""
        if self._confirm & ShellCmd.RM:
            answer = ConfirmEnum.NO
        else:
            answer = ConfirmEnum.ALL
        for script in scripts:
            if answer is not ConfirmEnum.ALL:
                answer = self.confirm(f'Remove {script}?', multi=True)
                if answer is ConfirmEnum.NONE:
                    self.logger.info('stopped')
                    break
            if answer:
                self.manager.deletescript(script)
            else:
                self.logger.info('keeping %s', script)

    def do_sh(self, *args: str):
        """sh [command] [argument ...] - run system command or shell"""
        if not args:
            args = (pwd.getpwuid(os.getuid()).pw_shell,)
        subprocess.run(args, check=True)   # nosec B603

    def do_su(self, user: str):
        """su user - manage scripts of user."""
        mgr = self.manager
        login = mgr.login
        if mgr.login:
            mgr.unauthenticate()
        # Some ManageSieve servers reject the first "AUTHENTICATE"
        # after an "UNAUTHENTICATE" regardless.
        for i in range(2):
            try:
                mgr.authenticate(login, *mgr.auth,
                                 owner=user, sasl=mgr.sasl, **mgr.kwauth)
            except OperationFailedError as err:
                if i:
                    raise err
                continue
            break

    def do_vi(self, *args: str):
        """vi [-a] script [...] - edit scripts with a visual editor"""
        mgr = self.manager
        (opts, scripts) = getopt(list(args), 'a')
        for opt, _ in opts:
            if opt == '-a':
                scripts = [script] if (script := mgr.getactive()) else []
        if scripts:
            editor = os.getenv('VISUAL', 'vi')
            mgr.editscripts([editor], list(scripts))
        else:
            self.logger.error('no scripts given')

    # Globbing and tab-completion
    @staticmethod
    def complete_dirnames(_argidx: int, text: str) -> list[str]:
        """Complete local directory names."""
        return list(readdir(path.dirname(text), path.isdir))

    @staticmethod
    def completefilenames(_argidx: int, text: str) -> list[str]:
        """Complete local filenames."""
        return list(readdir(path.dirname(text), path.isfile))

    def complete_scriptnames(self, _argidx: int, _text: str) -> list[str]:
        """Complete script names."""
        return [name for name, _ in self.manager.listscripts(cached=True)]

    complete_activate = complete_scriptnames
    """Completer for activate."""

    complete_cat = complete_scriptnames
    """Completer for cat."""

    complete_cd = complete_dirnames
    """Completer for cd."""

    complete_cmp = complete_scriptnames
    """Completer for cmp."""

    complete_cp = complete_scriptnames
    """Completer for cp."""

    complete_check = completefilenames
    """Completer for check."""

    complete_diff = complete_scriptnames
    """Completer for diff."""

    complete_ed = complete_scriptnames
    """Completer for ed."""

    def complete_get(self, argidx: int, text: str) -> list[str]:
        """Completer for get."""
        if argidx == 1:
            return self.complete_scriptnames(argidx, text)
        if argidx == 2:
            return self.completefilenames(argidx, text)
        return []

    def complete_help(self, _argidx: int, _text: str) -> tuple[str, ...]:
        """Completer for help."""
        return self.commands

    complete_ls = complete_scriptnames
    """Completer for ls."""

    complete_mget = complete_scriptnames
    """Completer for mget."""

    complete_more = complete_scriptnames
    """Completer for more."""

    complete_mput = completefilenames
    """Completer for mput."""

    complete_mv = complete_scriptnames
    """Completer for mv."""

    def complete_put(self, argidx: int, text: str) -> list[str]:
        """Completer for put."""
        if argidx == 1:
            return self.completefilenames(argidx, text)
        if argidx == 2:
            return self.complete_scriptnames(argidx, text)
        return []

    complete_rm = complete_scriptnames
    """Completer for rm."""

    complete_vi = complete_scriptnames
    """Completer for vi."""

    # Properties
    manager: SieveManager
    """Connection to a ManageSieve server."""


class InteractiveProxy(dict):
    """Proxy to use with :func:`code.interact`.

    Arguments:
        obj: Object to proxy.
    """

    def __init__(self, obj: Any):
        """Initialise a proxy."""
        super().__init__()
        self.obj = obj

    @staticmethod
    def exit():
        """Exit the Python read-evaluate-print loop."""
        raise SystemExit()

    def keys(self):
        """Same as :meth:`dir`."""
        return dir(self.obj) + list(globals().keys())

    def __getitem__(self, key):
        """Get an attribute of the proxied object."""
        try:
            return getattr(self, key)
        except AttributeError:
            pass
        try:
            return getattr(self.obj, key)
        except AttributeError:
            pass
        return globals()[key]


def columnise(words: Sequence[str]):
    """Print `words` in columns."""
    nwords = len(words)
    if nwords > 0:
        colwidth = max(map(len, words)) + 1
        maxncols = shutil.get_terminal_size((25, 80)).columns // colwidth
        nrows = math.ceil(nwords / maxncols)
        table: tuple[list[str], ...] = tuple([] for _ in range(nrows))
        for i, word in enumerate(words):
            table[i % nrows].append(word)
        for row in table:
            print(''.join([col.ljust(colwidth) for col in row]))


def readdir(dirname: str, predicate: Callable[[str], bool]) -> Iterator[str]:
    """Get every filename in `dirname` that matches `predicate`."""
    for dirent in os.listdir(dirname if dirname else '.'):
        fname = path.join(dirname, dirent)
        if predicate(fname):
            yield fname


#
# Configuration
#

class AttrDict():
    """Access attribute as dictionary items.

    For example:

    >>> obj = AttrDict()
    >>> obj.foo = 'bar'
    >>> obj['foo']
    'bar'

    Methods and private properties are ignored.
    """

    def __getitem__(self, key: str) -> Any:
        """Get an attribute."""
        try:
            return getattr(self, key)
        except AttributeError as err:
            raise KeyError(key) from err

    def __iter__(self) -> Iterator[Any]:
        """Iterate over attribute names."""
        return self.keys()

    def get(self, key: str, default: Any = None) -> Any:
        """Get the value of an attribute, but fallback to a default."""
        try:
            return getattr(self, key)
        except AttributeError:
            pass
        return default

    def keys(self) -> Iterator[Any]:
        """Get attribute names."""
        for key in dir(self):
            if not key.startswith('_') and not callable(self.get(key)):
                yield key

    def items(self) -> Iterator[tuple[Any, Any]]:
        """Get name-value pairs."""
        for key in self.keys():
            yield (key, self.get(key))

    def values(self) -> Iterator[Any]:
        """Get attribute values."""
        for _, value in self.items():
            yield value


class AttrProxyStore(AttrDict):
    """Back-end for proxied attributes.

    .. seealso::
        * :class:`AttrProxy`
    """

    def __init__(self, *args, **kwargs):
        """Initialise a new attribute proxy back-end.

        Arguments are used to populate the back-end with values.

        Arguments:
            args: Passed to `dict` as positional arguments.
            kwargs: Passed to `dict` as keyword arguments.
        """
        self._store = dict(*args, **kwargs)

    def __or__(self, other):
        """Merge two :class:`AttrProxyStore` objects."""
        ret = self.__class__()
        ret._store = self._store | other._store
        return ret

    def __ior__(self, other):
        """Merge :class:`AttrProxyStore` objects in-place."""
        self._store |= other._store
        return self

    def __ror__(self, other):
        """Merge two :class:`AttrProxyStore` objects in reverse."""
        ret = self.__class__()
        ret._store = other._store | self._store
        return ret

    def __setitem__(self, key: str, value: str):
        """Set `key` to `value`.

        :raises KeyError: Bad attribute.
        """
        if key.startswith('_'):
            raise KeyError(key)
        try:
            attr = getattr(self, key)
        except AttributeError as err:
            raise KeyError(key) from err
        if callable(attr):
            raise KeyError(key)
        setattr(self, key, value)

    def isset(self, name: str) -> bool:
        """Check if the `name` attribute has been set."""
        if name not in dir(self):
            raise AttributeError(f'{name}: no such attribute')
        try:
            return self._store[name] is not None
        except KeyError:
            return False

    def _get(self, name):
        """Get a value from the back-end."""
        return self._store[name]

    def _set(self, name, value):
        """Set a value in the back-end."""
        self._store[name] = value


class AttrProxy(ABC):
    """Base class for attribute proxies.

    Allows to mangle an attribute's value when it is set or read.
    Can only be used with :class:`AttrProxyStore` subclasses.

    For example:

    >>> class Foo(AttrProxyStore):
    >>>     foo = BoolAttrProxy('foo', default=False)
    >>>     bar = IntAttrProxy('bar', default=0)
    >>>
    >>> foo = Foo(bar=1)
    >>> foo.foo
    False
    >>> foo.bar
    1
    >>> foo.foo = 'yes'
    >>> foo.foo
    True
    >>> foo.bar = '2'
    >>> foo.bar
    2
    """

    def __init__(self, name: str, default: Any = None):
        """Initialise a proxied attribute.

        Arguments:
            name: Attribute name.
            default: Default.
        """
        self.name = name
        self.default = default

    def __get__(self, obj: AttrProxyStore, _cls: type) -> Any:
        """Get the attribute's value.

        Arguments:
            obj: Object to which the attribute belongs.

        Raises:
            ValueError: `value` is invalid.
        """
        try:
            return obj._get(self.name)
        except KeyError:
            return self.default

    def __set__(self, obj: AttrProxyStore, value: Optional[str]):
        """Set the attribute's value.

        Arguments:
            obj: Object to which the attribute belongs.
            value: Value to mangle.

        Raises:
            ValueError: `value` is invalid.
        """
        obj._set(self.name, self.default if value is None else value)


class ManglingAttrProxy(AttrProxy):
    """Base class for attribute proxies that mangle values."""

    @abstractmethod
    def mangle(self, obj: AttrProxyStore, value: Any) -> Any:
        """Mangle `value`.

        Arguments:
            obj: Object to which the attribute belongs.
            value: Value to mangle.

        Raises:
            ValueError: `value` is invalid.
        """


class ListAttrProxyTrait():
    """Trait for lists."""

    _splitregex = re.compile(r'\s*,\s*')
    """Regular expression that splits comma-separated lists."""


class EagerAttrProxy(ManglingAttrProxy):
    """Base class for 'eager' attribute proxies (mangled on write)."""

    def __set__(self, obj: AttrProxyStore, value: Any):
        if value == '':
            raise ValueError(f'{self.name}: empty')
        super().__set__(obj, self.mangle(obj, value))


class LazyAttrProxy(ManglingAttrProxy):
    """Base class for 'lazy' attribute proxies (mangled on read)."""

    def __get__(self, obj: AttrProxyStore, cls: type) -> Any:
        return self.mangle(obj, super().__get__(obj, cls))


class ExpAttrProxy(LazyAttrProxy):
    """Expand '~', environment variables, and attributes."""

    def expand(self, obj: AttrProxyStore, value: str) -> str:
        """Expand '~', environment variables, and attributes."""
        name = self.name
        template = string.Template(value)
        # template.get_identifiers is only available in Python >= v3.11.
        varnames = set()
        # pylint: disable=consider-using-f-string
        for pattern in (r'\$(%s)' % template.idpattern,
                        r'\$\{(%s)\}' % template.idpattern):
            for match in re.finditer(pattern, value, flags=re.IGNORECASE):
                varnames.add(match.group(1))
        vars_ = {}
        for varname in varnames:
            try:
                value = getattr(obj, varname)
            except AttributeError as err:
                raise ValueError(f'{name}: ${varname}: no such key') from err
            if value is None:
                raise ValueError(f'{name}: ${varname}: not set')
            if not isinstance(value, (int, str)):
                raise ValueError(f'{name}: ${varname}: not a scalar.')
            vars_[varname] = value
        return path.expanduser(template.substitute(vars_))

    def mangle(self, obj: AttrProxyStore,
               value: Optional[str]) -> Optional[str]:
        return None if value is None else self.expand(obj, value)


class NumAttrProxy(EagerAttrProxy):
    """Convert attribute to a number of type :attr:`cls`."""

    # pylint: disable=redefined-builtin
    def __init__(self, *args,
                 min: Optional[Union[int, float]] = None,
                 max: Optional[Union[int, float]] = None,
                 **kwargs):
        """Initialise the proxy.

        Arguments:
            name: Attribute name.
            min: Smallest permissible value.
            max: Greatest permissible value.
            default: Default.
        """
        super().__init__(*args, **kwargs)
        self.min = min
        self.max = max

    def mangle(self, _obj: Any,
               value: Union[int, float, str]) -> Union[int, float]:
        try:
            num = self.cls(value)
        except ValueError as err:
            raise ValueError(f'{self.name}: {value}: not a number') from err
        if self.min is not None and num < self.min:
            raise ValueError(f'{self.name}: {value} < {self.min}')
        if self.max is not None and num > self.max:
            raise ValueError(f'{self.name}: {value} > {self.max}')
        return num

    @property
    @abstractmethod
    def cls(self) -> type:
        """Numeric type to convert values to."""


class AuthAttrProxy(EagerAttrProxy, ListAttrProxyTrait):
    """Convert SASL mechanism names to objects."""

    def mangle(self, obj: AttrProxyStore,
               value: Union[Iterable[type[BaseAuth]], str]) \
                    -> list[type[BaseAuth]]:
        if isinstance(value, str):
            classes = BaseAuth.subclasses()
            mechs = []
            for name in self._splitregex.split(value.casefold()):
                matches = []
                for cls in classes:
                    if fnmatch.fnmatchcase(cls.name.casefold(), name):
                        if cls in mechs:
                            message = f'{self.name}: {cls.name}: duplicate'
                            raise ValueError(message)
                        matches.append(cls)
                if not matches:
                    raise ValueError(f'{self.name}: {name}: no matches')
                mechs.extend(matches)
        elif isinstance(value, Sequence):
            mechs = value     # type: ignore[assignment]
        if len(set(m.type for m in mechs)) > 1:
            raise ValueError(f'{self.name}: bad combination')
        return mechs

    superclasses = (BasePassAuth, ExternalAuth)
    """Authentication types."""


class BoolAttrProxy(EagerAttrProxy):
    """Convert "yes" and "no" to :class:`bool`."""

    def mangle(self, _obj: Any, value: Union[bool, str]) -> bool:
        if value in (True, 'yes'):
            return True
        if value in (False, 'no'):
            return False
        raise ValueError(f'{self.name}: {value}: not a boolean')


class CmdAttrProxy(ExpAttrProxy):
    """Split up attribute into a list using :func:`shlex.split`."""

    def mangle(self,    # type: ignore[override]
               obj: AttrProxyStore, value: Optional[str]) -> Optional[list]:
        if value is not None:
            return [self.expand(obj, word)
                    for word in shlex.split(value, posix=True)]
        return None


class EnumAttrProxy(EagerAttrProxy):
    """Convert attribute to an :class:`enum.Enum`."""

    def __init__(self, *args, cls: type[enum.Enum], **kwargs):
        """Initialise the proxy.

        Arguments:
            name: Attribute name.
            cls: Type to convert the value to.
            default: Default.
        """
        assert issubclass(cls, enum.Enum)    # nosec B101
        super().__init__(*args, **kwargs)
        self.cls = cls

    def mangle(self, obj: AttrProxyStore,
               value: Optional[Union[enum.Enum, str]]) -> Optional[enum.Enum]:
        if value is None:
            return None
        if isinstance(value, enum.Enum):
            return value
        for member in self.cls:
            if member.name.casefold() == value.casefold():
                return member
        raise ValueError(f'{self.name}: {value}: no such item')


class FlagAttrProxy(EnumAttrProxy, ListAttrProxyTrait):
    """Convert attribute to an :class:`int`."""

    def mangle(self, obj: AttrProxyStore,   # type: ignore[override]
               value: Optional[str]) -> int:
        flag = 0
        if value is not None:
            for name in self._splitregex.split(value):
                for member in self.cls:
                    if name.casefold() == member.name.casefold():
                        flag |= member.value
                        break
                else:
                    raise ValueError(f'{self.name}: {name}: no such item')
        return flag


class FloatAttrProxy(NumAttrProxy):
    """Convert attribute to a :class:`float`."""

    cls = float


class GlobalAttrProxy(BoolAttrProxy):
    """Set global variable when attribute is changed."""

    def __init__(self, *args, var: str, **kwargs):
        """Initialise the proxy.

        Arguments:
            name: Attribute name.
            default: Default.
            var: Variable name.
        """
        super().__init__(*args, **kwargs)
        self.var = var

    def mangle(self, obj: AttrProxyStore,
               value: str) -> bool:     # type: ignore[override]
        boolean = super().mangle(obj, value)
        globals()[self.var] = boolean
        return boolean


class HostAttrProxy(EagerAttrProxy):
    """Check whether value is a valid hostname."""

    def mangle(self, _obj: AttrProxyStore,
               value: Optional[str]) -> Optional[str]:
        if value is None:
            return self.default
        if (match := self._hostregex.fullmatch(value)) is not None:
            for host in match.groupdict().values():
                if host is not None:
                    return host
        raise ValueError(f'{self.name}: {value}: neither hostname nor address')

    _hostregex = re.compile('|'.join((
        r'(?P<ipv4>([0-2]?[0-9]{1,2}\.){3}[0-2]?[0-9]{1,2})',
        r'\[(?P<ipv6>([0-9a-f]{0,4}:){2,7}[0-9a-f]{0,4})\]',
        r'(?P<hostname>([a-z0-9]([a-z0-9-]{0,61}[a-z0-9]\.)?)+)',
    )), flags=re.IGNORECASE)
    """Matches hosts."""


class IntAttrProxy(NumAttrProxy):
    """Convert attribute to an integer."""

    cls = int


class Config(AttrProxyStore):
    """Configuration for the SieveManager command-line client."""

    @classmethod
    def fromfile(cls, fname: str) -> 'Config':
        """Read configuration from `fname`.

        See :doc:`docs/sieverc.rst` for the syntax.

        :raise ConfigError: Syntax error.
        """
        conf = cls()
        ptr = conf
        with open(fname, encoding='utf8') as file:
            for i, line in enumerate(file.readlines(), start=1):
                if line == '\n' or cls._commentregex.match(line):
                    continue
                try:
                    key, value = line.rstrip().split(maxsplit=1)
                    if key == 'account':
                        try:
                            login, host = value.rsplit('@', maxsplit=1)
                            ptr = cls(host=host, login=login)
                            ptr._login = login
                            ptr._host = host
                        except ValueError:
                            ptr = cls(host=value)
                        conf._sections.append(ptr)
                    else:
                        ptr[key] = value
                except KeyError as err:
                    message = f'{fname}:{i}: {key}: no such variable'
                    raise ConfigError(message) from err
                except (ValueError, Error) as err:
                    raise ConfigError(f'{fname}:{i}: {err}') from err
        conf.confdir = path.dirname(fname)
        return conf

    def __init__(self, *args: Any, **kwargs: Any):
        super().__init__(*args, **kwargs)
        self._login = None
        self._sections = []

    def __or__(self, other):
        """Merge two configurations."""
        conf = super().__or__(other)
        conf._sections = self._sections + other._sections
        return conf

    def __ior__(self, other):
        """Merge two configurations in-place."""
        super().__ior__(other)
        self._sections += other._sections
        return self

    def __ror__(self, other):
        """Merge two configurations in reverse."""
        conf = super().__ror__(other)
        conf._sections = other._sections + self._sections
        return conf

    def __repr__(self):
        """Get a string representation of the configuration."""
        variables = ', '.join((f'{k}={v!r}' for k, v in self.items()))
        return f'{self.__class__.__name__}({variables})'

    def getsection(self, host: str, login: Optional[str] = None) -> 'Config':
        """Get the configuration for `login` on `host`."""
        conf = self.__class__() | self
        for sec in self._sections:
            # pylint: disable=protected-access
            if host in (sec._host, sec.alias) and sec._login in (None, login):
                conf |= sec
        return conf

    alias = AttrProxy('alias')
    """Alias for a host."""

    authmechs = AuthAttrProxy('authmechs', BasePassAuth.subclasses())
    """How to authenticate."""

    backup = EnumAttrProxy('backup', Backup.NONE, cls=Backup)
    """Backup policy."""

    cadir = ExpAttrProxy('cadir')
    """Custom CA directory."""

    cafile = ExpAttrProxy('cafile')
    """Custom CA file."""

    confdir = ExpAttrProxy('confdir', f'{XDGCONFIGDIR}/sieve')
    """Configuration directory."""

    confirm = FlagAttrProxy('confirm', ShellCmd.ALL, cls=ShellCmd)
    """Which shell commands have to be confirmed?"""

    cert = ExpAttrProxy('cert')
    """Client TLS certificate."""

    debugauth = BoolAttrProxy('debugauth', False)
    """Log the authentication exchange?"""

    getkeypass = CmdAttrProxy('getkeypass')
    """Command that prints the passphrase for the TLS key."""

    getpass = CmdAttrProxy('getpass')
    """Command that prints a password."""

    host = HostAttrProxy('host', 'localhost')
    """Host to connect to by default."""

    key = ExpAttrProxy('key')
    """Client TLS key."""

    login = AttrProxy('login')
    """Login name to use by default."""

    ocsp = BoolAttrProxy('ocsp', True)
    """Check whether server certificate was revoked?"""

    port = IntAttrProxy('port', 4190, min=0, max=65535)
    """Port to connect to by default."""

    saslprep = FlagAttrProxy('saslprep', SASLPrep.ALL, cls=SASLPrep)
    """Which credentials to prepare."""

    stacktrace = GlobalAttrProxy('stacktrace', False, var='DEBUG')
    """Show stack traces?"""

    timeout = FloatAttrProxy('timeout', socket.getdefaulttimeout(), min=0)
    """Network timeout."""

    tls = BoolAttrProxy('tls', True)
    """Use TLS?"""

    verbosity = EnumAttrProxy('verbosity', LogLevel.INFO, cls=LogLevel)
    """Logging level."""

    x509strict = BoolAttrProxy('x509strict', True)
    """Be strict when verifying TLS certificates?"""

    _commentregex = re.compile(r'\s*#')
    """Controls which lines are considered comments."""

    _host: Optional[str] = None
    """Section host (if this is a section)."""

    _login: Optional[str] = None
    """Section login (if this is a section)."""

    _sections: list['Config'] = []
    """Configuration sections (if this is the global configuartion)."""


#
# Password stores
#

class BasePasswordStore(ABC):
    """Base class for retrieving credentials from password stores."""

    @abstractmethod
    def getlogin(self, host: str) -> Optional[str]:
        """Get a login name for `host`."""

    @abstractmethod
    def getpass(self, host: str, login: str) -> Optional[str]:
        """Get the password for `login` on `host`."""


class ConfigPasswordStore(BasePasswordStore):
    """Read credentials from a SieveManager configuration file."""

    def __init__(self, conf: 'Config'):
        """Initialise a :class:`ConfigPasswordStore` object."""
        self.conf = conf

    def getlogin(self, host: str) -> Optional[str]:
        conf = self.conf | self.conf.__class__(host=host)
        return conf.login

    def getpass(self, host: str, login: str) -> Optional[str]:
        conf = self.conf | self.conf.__class__(host=host, login=login)
        if getpass := conf.getpass:
            try:
                return readexec(*getpass)
            except subprocess.CalledProcessError:
                pass
        return None


class NetRC(netrc.netrc, BasePasswordStore):
    """Read credentials from .netrc

    .. seealso::
        * GNU `Inetutils manual`_ (chap. 11.7)

    .. _`Inetutils manual`: https://www.gnu.org/software/inetutils/manual
    """

    def getlogin(self, host: str) -> Optional[str]:
        try:
            return self.hosts[host][0]
        except KeyError:
            return None

    def getpass(self, host: str, login: str) -> Optional[str]:
        try:
            return self.hosts[host][2]
        except KeyError:
            return None


class PasswordStoreList(list, BasePasswordStore):
    """Read credentials from multiple password stores."""

    def getlogin(self, host: str):
        for store in self:
            if login := store.getlogin(host):
                return login
        return None

    def getpass(self, host: str, login: str):
        for store in self:
            if passwd := store.getpass(host, login):
                return passwd
        return None


def readexec(*command: str):
    """Read the output of `command`."""
    logging.debug('exec: %s', ' '.join(command))
    try:
        retval = subprocess.run(command,   # nosec B603
                                capture_output=True, check=True)
    except subprocess.CalledProcessError as err:
        logging.debug('%s: exited with status %d', err.cmd[0], err.returncode)
        raise err
    return retval.stdout.rstrip().decode(locale.getpreferredencoding())


#
# Errors
#

class Error(Exception):
    """Base class for errors."""


class AskpassError(Error):
    """Cannot ask for a password."""


class ConfigError(Error):
    """Configuration error."""


class CertRevokedError(Error):
    """Server certificate may have been revoked."""


class HTTPError(Error):
    """HTTP GET request failed."""


class OCSPError(Error):
    """Server certificate was revoked."""


class OperationError(Error):
    """Operation could not be performed or failed."""


class ProtocolError(Error):
    """Server has violated the ManageSieve protocol.

    .. warning::
        Continuing after a :exc:`ProtocolError` is unsafe.
    """


class UsageError(Error):
    """Usage error."""


class VerificationError(Error):
    """Server identity could not be verified."""


class ResolutionError(ConnectionError, Error):
    """Name could not be resolved."""


class SocketError(ConnectionError, Error):
    """Socket error."""


class ConnectionClosedError(Response, ConnectionError, Error):
    """Server said "BYE"."""

    # pylint: disable=redefined-outer-name
    def __init__(self,
                 response: str = 'bye',
                 code: str = '',
                 data: tuple[Union[int, str], ...] = (),
                 message: Optional[str] = None):
        super().__init__(response, code, data, message)


class ConnectionOpenError(ConnectionError, Error):
    """Connection already established."""


class CapabilityError(OperationError):
    """Capability not supported by the server."""


class AuthMechError(CapabilityError):
    """The server does not support any of the given SASL mechanisms."""


class ProxyAuthError(CapabilityError):
    """Authorisation not supported by the selected SASL mechanism."""


class TLSCapabilityError(CapabilityError):
    """The server does not support "STARTTLS"."""


class OperationOngoingError(OperationError):
    """Another operation is in progress."""


class OperationFailedError(Response, OperationError):
    """Server said "NO"."""

    # pylint: disable=redefined-outer-name
    def __init__(self,
                 response: str = 'no',
                 code: str = '',
                 data: tuple[Union[int, str], ...] = (),
                 message: Optional[str] = None):
        super().__init__(response, code, data, message)


@dataclass(frozen=True)
class SignalCaught(Exception):
    """Signal was caught."""

    @classmethod
    def catch(cls, signo: int, frame: Optional[types.FrameType]):
        """Raise a :exc:`SignalCaught`."""
        raise cls(signo, frame)

    def __str__(self):
        desc = signal.strsignal(self.signo)
        return desc.split(':')[0] if desc else f'caught signal {self.signo}'

    signo: int
    """Signal number."""

    frame: Optional[types.FrameType] = None
    """Stack frame."""


#
# Main
#

def askpass(prompt: str = 'Password: ') -> str:
    """Prompt for a password."""
    if os.isatty(sys.stdin.fileno()):
        return getpass.getpass(prompt)
    if platform.system() == 'Darwin':
        applescript = """on run argv
            set text item delimiters to " "
            set prompt to argv as text
            set reply to display dialog prompt ¬
                default answer "" with title "SieveManager" with hidden answer
            set pass to text returned of reply
            return pass
        end run"""
        return readexec('osascript', '-e', applescript, prompt)
    if 'tkinter' in sys.modules:
        dialogue = tkinter.Tk()
        dialogue.title('SieveManager')
        field = tkinter.Frame(dialogue)
        field.pack(padx=5, pady=(5, 0))
        label = tkinter.Label(field, text=prompt.rstrip())
        label.grid(row=1, column=1)
        entry = tkinter.Entry(field, show="*")
        entry.grid(row=1, column=2)
        buttons = tkinter.Frame(dialogue)
        buttons.pack(padx=5, pady=(0, 5))
        ok = tkinter.Button(buttons, text='OK', command=dialogue.quit)
        ok.grid(row=2, column=1)
        cancel = tkinter.Button(buttons, text='Cancel', command=exit)
        cancel.grid(row=2, column=2)
        dialogue.mainloop()
        password = entry.get()
        dialogue.destroy()
        return password
    raise AskpassError('password needed, but standard input is not a TTY')


def findconfigfile(basename: str) -> Optional[str]:
    """Find a configuration file.

    Looks in:
        * :samp:`$XDG_CONFIG_HOME/{basename}/{basename}rc`
        * :samp:`$HOME/.{basename}/{basename}rc`
        * :samp:`$HOME/.{basename}rc`
        * :samp:`/etc/{basename}/{basename}rc`
        * :samp:`/etc/{basename}rc`
    """
    fnames = (f'{XDGCONFIGDIR}/{basename}/{basename}rc',
              f'{HOMEDIR}/.{basename}/{basename}rc',
              f'{HOMEDIR}/.{basename}rc',
              f'/etc/{basename}/{basename}rc',
              f'/etc/{basename}rc')
    for fname in fnames:
        if path.exists(fname):
            return fname
    return None


# pylint: disable=too-many-branches, too-many-locals, too-many-statements
def sievemgr():
    """sievemgr - manage Sieve scripts remotely

    Usage:  sievemgr [-c file] [-f] [-o key=value] [-q|-v]
                     [server] [command] [argument ...]

    Operands:
        server        ManageSieve server to connect to (default: localhost).
        command       Command to run.
        argument      Argument to that command.

    Options:
        -c file       Read configuration from file.
        -d            Enable debugging mode.
        -f            Overwrite and remove files without confirmation.
        -o key=value  Set configuration key to value.
        -q            Be quieter. Can be given multiple times.
        -v            Be more verbose. Can be given multiple times.

        See sievemgr(1) for the complete list.

    Home page: <https://odkr.codeberg.page/sievemgr>
    """
    progname = path.basename(sys.argv[0])
    logging.basicConfig(format=f'{progname}: %(message)s')

    for signo in (SIGHUP, SIGINT, SIGTERM):
        signal.signal(signo, SignalCaught.catch)

    # Options
    configfile = findconfigfile('sieve')
    confoverride = Config()
    logdelta = 0
    netrcfile = os.getenv('NETRC')
    opts, args = getopt(sys.argv[1:], 'N:Vdc:fho:qv', ['help', 'version'])
    for opt, arg in opts:
        try:
            if opt in ('-h', '--help'):
                assert sievemgr.__doc__     # nosec B101
                for line in sievemgr.__doc__.splitlines()[:-1]:
                    print(re.sub(r'^ {4}', '', line))
                sys.exit()
            elif opt in ('-V', '--version'):
                print(f'SieveManager {__version__}\nCopyright {__copyright__}')
                sys.exit()
            elif opt == '-N':
                netrcfile = arg
            elif opt == '-c':
                configfile = arg
            elif opt == '-d':
                confoverride.verbosity = LogLevel.DEBUG
                confoverride.stacktrace = True
            elif opt == '-f':
                confoverride.confirm = ShellCmd.NONE
            elif opt == '-o':
                try:
                    key, value = arg.split('=', maxsplit=1)
                except ValueError as err:
                    key = arg
                    if isinstance(confoverride[key], bool):
                        value = True    # type: ignore[assignment]
                    else:
                        raise ValueError(f'{key}: no value') from err
                if value == '':
                    raise ValueError(f'{key}: empty')
                confoverride[key] = value
            elif opt == '-q':
                logdelta -= 1
            elif opt == '-v':
                logdelta += 1
        except KeyError as err:
            raise UsageError(f'{opt}: {key}: no such key') from err
        except (Error, ValueError) as err:
            raise UsageError(f'{opt}: {err}') from err

    # Preliminary configuration
    conf = ((Config.fromfile(configfile) | confoverride) if configfile
            else confoverride)

    # Arguments
    try:
        url = URL.fromstr(args[0])
    except IndexError:
        url = URL(hostname=conf.host)
    try:
        command = shlex.join(args[1:])
    except IndexError:
        command = None

    # Final configuration
    conf = conf.getsection(url.hostname, url.username) | confoverride

    # Logging level
    levels = list(LogLevel)
    logindex = levels.index(conf.verbosity) - logdelta
    level = levels[min(max(logindex, 0), len(levels) - 1)]
    logging.getLogger().setLevel(level)
    logauth = level <= LogLevel.AUTH

    # Infos
    logging.info('SieveManager %s', __version__)
    logging.info('%s', __copyright__)
    logging.info('This application comes with ABSOLUTELY NO WARRANTY.')

    # Password stores
    credentials = PasswordStoreList([ConfigPasswordStore(conf)])
    try:
        if netrcfile:
            credentials.append(NetRC(netrcfile))
        else:
            try:
                credentials.append(NetRC())
            except FileNotFoundError:
                pass
    except netrc.NetrcParseError as err:
        if sys.version_info < (3, 10):
            logging.error(err)
        else:
            raise err

    # TLS
    if conf.isset('cadir') or conf.isset('cafile'):
        SieveManager.sslcontext.load_verify_locations(conf.cafile, conf.cadir)
    if cert := conf.cert:
        def getpassphrase():
            if getkeypass := conf.getkeypass:
                return readexec(*getkeypass)
            return askpass('Certificate passphrase: ')
        SieveManager.sslcontext.load_cert_chain(cert, conf.key, getpassphrase)
    if conf.x509strict:
        SieveManager.sslcontext.verify_flags |= ssl.VERIFY_X509_STRICT

    # Connection and authentication information
    host = conf.host if conf.alias == url.hostname else url.hostname
    port = url.port if url.port else conf.port
    login = url.username or credentials.getlogin(host) or getpass.getuser()
    owner = url.owner if url.owner else ''
    sasl = conf.authmechs

    # Connect and start the shell
    with SieveManager(host, '', port=port, timeout=conf.timeout,
                      tls=conf.tls, ocsp=conf.ocsp,
                      backup=conf.backup) as mgr:
        if issubclass(sasl[0], BasePassAuth):
            password = (url.password
                        or credentials.getpass(host, login)
                        or askpass())
            mgr.authenticate(login, password, owner=owner, sasl=sasl,
                             logauth=logauth, prepare=conf.saslprep)
        elif issubclass(sasl[0], ExternalAuth):
            mgr.authenticate(login, owner=owner, sasl=sasl,
                             logauth=logauth, prepare=conf.saslprep)
        else:
            raise NotImplementedError('unsupported authentication type')
        shell = SieveShell(mgr, confirm=conf.confirm)
        if command:
            shell.run(command)
        else:
            shell.enter()


def main():
    """Call :func:`sievemgr` but catch exceptions."""
    try:
        sievemgr()
    except SignalCaught as exc:
        logging.error(exc)
        signal.signal(SIGTERM, SIG_IGN)
        os.killpg(os.getpgrp(), SIGTERM)
        signal.signal(exc.signo, SIG_DFL)
        signal.raise_signal(exc.signo)
    except (GetoptError, UsageError) as err:
        logging.error(err)
        sys.exit(2)
    # pylint: disable=broad-exception-caught
    except Exception as err:
        if DEBUG:
            raise err
        if isinstance(err, (socket.herror, socket.gaierror, SocketError)):
            logging.error('%s', err.args[1])
        elif isinstance(err, (FileExistsError, FileNotFoundError)):
            logging.error('%s: %s', err.filename, os.strerror(err.errno))
        elif isinstance(err, (MemoryError, ssl.SSLError, Error)):
            logging.error('%s', err)
        elif isinstance(err, OSError):
            logging.error('%s', os.strerror(err.errno))
        elif isinstance(err, subprocess.CalledProcessError):
            logging.error('%s: exited with status %d',
                          err.cmd[0], err.returncode)
        elif 'tkinter' in sys.modules and isinstance(err, tkinter.TclError):
            logging.error('%s', err)
        else:
            traceback.print_exc(file=sys.stderr)
        sys.exit(1)


#
# Logging
#

logging.getLogger(__name__).addHandler(logging.NullHandler())


#
# Boilerplate
#

if __name__ == '__main__':
    main()
